الگوی Saga

۷.۱ مقدمه

یکی از پیچیده‌ترین چالش‌های میکروسرویس، مدیریت تراکنش‌هایی است که چندین سرویس را شامل می‌شوند. الگوی Saga راه‌حل اصلی این مشکل است.

۷.۲ مشکل: Distributed Transactions

تراکنش در Monolith

در یک Monolith با یک database، تراکنش ساده است:

@transaction.atomic
def place_order(user_id, items):
    # همه در یک تراکنش — یا همه موفق، یا هیچ‌کدام
    order = Order.objects.create(user_id=user_id)
    for item in items:
        OrderItem.objects.create(order=order, **item)
        Inventory.objects.filter(product_id=item["product_id"]).update(
            stock=F("stock") - item["qty"]
        )
    Payment.objects.create(order=order, amount=order.total)
    Wallet.objects.filter(user_id=user_id).update(
        balance=F("balance") - order.total
    )
    # commit خودکار اگر همه موفق
    # rollback خودکار اگر یکی fail
    

چالش در Microservice

هر سرویس DB مستقل خود را دارد. ACID transaction سراسری ممکن نیست!

ثبت سفارش = چندین سرویس باید با هم کار کنند:

1. Order Service: ساخت Order
2. Inventory Service: کاهش موجودی
3. Payment Service: کسر از کیف پول
4. Shipping Service: ایجاد shipment
5. Notification Service: ارسال ایمیل

اگر مرحله ۳ (پرداخت) fail شود:
- مراحل ۱ و ۲ قبلاً انجام شده‌اند
- چطور rollback کنیم؟
    

راه‌حل‌های ممکن

• Two-Phase Commit (2PC): کند، blocking، single point of failure
• Three-Phase Commit (3PC): پیچیده‌تر، هنوز مشکلات داشت
• Saga Pattern: راه‌حل عملی و مدرن (انتخاب اول)

۷.۳ الگوی Saga

Saga یک تراکنش بزرگ را به دنباله‌ای از تراکنش‌های محلی (Local Transactions) تقسیم می‌کند. هر تراکنش محلی روی یک سرویس انجام می‌شود و یک event منتشر می‌کند که مرحله بعدی را trigger می‌کند.

اصول Saga

• هر مرحله یک تراکنش محلی است (ACID در سرویس خودش)
• اگر یکی شکست بخورد، تراکنش‌های قبلی با Compensation برگردانده می‌شوند
• Saga eventual consistency را پذیرفته است
• هیچ rollback خودکار نیست — همه چیز explicit است

مثال: ثبت سفارش

مسیر موفق (Happy Path):
    
1. CreateOrder ──────────────► OrderCreated
2. ReserveInventory ─────────► InventoryReserved
3. ChargePayment ────────────► PaymentCharged
4. CreateShipment ───────────► ShipmentCreated
5. SendConfirmation ─────────► OrderCompleted ✓

مسیر شکست (در مرحله ۳):
    
1. CreateOrder ──────────────► OrderCreated
2. ReserveInventory ─────────► InventoryReserved
3. ChargePayment ──────FAIL──► PaymentFailed
   ↓
4. ReleaseInventory ─────────► InventoryReleased   (compensation)
5. CancelOrder ──────────────► OrderCancelled       (compensation)

نتیجه: همه چیز برگشت به حالت اولیه ✓
    

۷.۴ Compensation Transactions

برای هر تراکنش، یک «compensation» تعریف می‌کنیم که اثر آن را خنثی می‌کند.

اصول طراحی Compensation

• Idempotent: اجرای چند بار باید همان نتیجه را بدهد
• Commutative نباشد: ترتیب compensation معمولاً معکوس ترتیب اصلی است
• هیچ‌گاه fail نشود: compensation باید تضمینی موفق شود (با retry)
• Semantically reversing: اثر اصلی را خنثی کند، نه اینکه فقط تغییر را پاک کند

مشکل: Compensation همیشه ممکن نیست!

مثلاً اگر ایمیل ارسال شد، نمی‌توان آن را «un-send» کرد. در این موارد، compensation معنایی متفاوت می‌گیرد:

SendOrderConfirmation ──fail later──► SendCancellationEmail
                                        (به جای un-send)
    

۷.۵ Choreography (هم‌نوازی)

در Choreography، هیچ مرکز کنترلی وجود ندارد. هر سرویس به event ها گوش می‌دهد و خودش می‌داند چه کار کند.

[Order Service]
     │
     │ publishes: OrderCreated
     ▼
[Event Bus] ──┬──► [Inventory Service] (listens, reserves)
              │       │
              │       │ publishes: InventoryReserved
              │       ▼
              │  [Event Bus] ──► [Payment Service] (listens, charges)
              │                       │
              │                       │ publishes: PaymentCharged
              │                       ▼
              │                  [Event Bus] ──► [Shipping] (creates shipment)
    

پیاده‌سازی Choreography

# Order Service
@app.post("/orders")
async def create_order(data: OrderCreate):
    async with db.transaction():
        order = await Order.create(
            user_id=data.user_id,
            items=data.items,
            status="pending",
        )
        # Event در outbox
        await publish_event("order.created", {
            "order_id": order.id,
            "user_id": order.user_id,
            "items": order.items,
            "total": order.total,
            "saga_id": str(uuid.uuid4()),
        })
    return order

# Inventory Service
@event_handler("order.created")
async def on_order_created(event):
    order_id = event["order_id"]
    items = event["items"]
    saga_id = event["saga_id"]
    
    try:
        async with db.transaction():
            for item in items:
                await Inventory.reserve(
                    product_id=item["product_id"],
                    qty=item["qty"],
                    saga_id=saga_id,
                )
        
        await publish_event("inventory.reserved", {
            "order_id": order_id,
            "saga_id": saga_id,
        })
    except InsufficientStockError as e:
        await publish_event("inventory.reservation_failed", {
            "order_id": order_id,
            "saga_id": saga_id,
            "reason": str(e),
        })

# Payment Service
@event_handler("inventory.reserved")
async def on_inventory_reserved(event):
    order_id = event["order_id"]
    saga_id = event["saga_id"]
    
    order = await fetch_order(order_id)
    
    try:
        payment = await charge_payment(
            user_id=order["user_id"],
            amount=order["total"],
            saga_id=saga_id,
        )
        await publish_event("payment.charged", {
            "order_id": order_id,
            "payment_id": payment.id,
            "saga_id": saga_id,
        })
    except PaymentFailedError as e:
        await publish_event("payment.failed", {
            "order_id": order_id,
            "saga_id": saga_id,
            "reason": str(e),
        })

# Order Service - listening to compensations
@event_handler("inventory.reservation_failed")
async def on_inventory_failed(event):
    await Order.update(
        order_id=event["order_id"],
        status="cancelled",
        cancellation_reason=event["reason"],
    )

@event_handler("payment.failed")
async def on_payment_failed(event):
    # نیاز به compensation: release inventory + cancel order
    await publish_event("inventory.release_request", {
        "order_id": event["order_id"],
        "saga_id": event["saga_id"],
    })

# Inventory Service - compensation
@event_handler("inventory.release_request")
async def on_release_request(event):
    saga_id = event["saga_id"]
    await Inventory.release_by_saga(saga_id)
    await publish_event("inventory.released", {
        "order_id": event["order_id"],
        "saga_id": saga_id,
    })
    

مزایا و معایب Choreography

کاربرد مناسب: Saga های ساده با ۲-۳ مرحله.

۷.۶ Orchestration (هدایت)

در Orchestration، یک سرویس مرکزی به نام «Orchestrator» (یا «Saga Manager») کل فرآیند را هدایت می‌کند.

        ┌─────────────────────────────┐
        │   Order Saga Orchestrator   │
        └─────────────────────────────┘
               │  ▲   │  ▲   │  ▲
               │  │   │  │   │  │
               ▼  │   ▼  │   ▼  │
        ┌──────────┐  ┌──────────┐  ┌──────────┐
        │  Order   │  │Inventory │  │ Payment  │
        │ Service  │  │ Service  │  │ Service  │
        └──────────┘  └──────────┘  └──────────┘

Orchestrator می‌گوید: "Order Service، یک سفارش بساز"
Order Service: "ساختم"
Orchestrator: "Inventory، کم کن"
Inventory: "OK"
... و الی آخر
    

پیاده‌سازی State Machine

# orchestrator/saga_state.py
from enum import Enum
from dataclasses import dataclass
from typing import Optional

class SagaStep(str, Enum):
    STARTED = "started"
    ORDER_CREATED = "order_created"
    INVENTORY_RESERVED = "inventory_reserved"
    PAYMENT_CHARGED = "payment_charged"
    SHIPMENT_CREATED = "shipment_created"
    COMPLETED = "completed"
    
    # Compensation states
    COMPENSATING = "compensating"
    INVENTORY_RELEASED = "inventory_released"
    PAYMENT_REFUNDED = "payment_refunded"
    ORDER_CANCELLED = "order_cancelled"
    FAILED = "failed"

@dataclass
class SagaContext:
    saga_id: str
    user_id: int
    items: list
    total: float
    
    # State
    current_step: SagaStep = SagaStep.STARTED
    order_id: Optional[int] = None
    payment_id: Optional[int] = None
    shipment_id: Optional[int] = None
    
    # Compensation tracking
    completed_steps: list = None
    failure_reason: Optional[str] = None
    
    def __post_init__(self):
        if self.completed_steps is None:
            self.completed_steps = []
    

Orchestrator اصلی

# orchestrator/order_saga.py
import asyncio
import logging
from .saga_state import SagaContext, SagaStep
from .clients import OrderClient, InventoryClient, PaymentClient, ShippingClient

logger = logging.getLogger(__name__)

class OrderSagaOrchestrator:
    def __init__(self):
        self.order_client = OrderClient()
        self.inventory_client = InventoryClient()
        self.payment_client = PaymentClient()
        self.shipping_client = ShippingClient()
    
    async def execute(self, ctx: SagaContext) -> SagaContext:
        """اجرای saga"""
        try:
            # Step 1: Create Order
            await self._create_order(ctx)
            
            # Step 2: Reserve Inventory
            await self._reserve_inventory(ctx)
            
            # Step 3: Charge Payment
            await self._charge_payment(ctx)
            
            # Step 4: Create Shipment
            await self._create_shipment(ctx)
            
            # Success
            ctx.current_step = SagaStep.COMPLETED
            await self._save_state(ctx)
            logger.info(f"Saga {ctx.saga_id} completed")
            
        except Exception as e:
            logger.error(f"Saga {ctx.saga_id} failed at {ctx.current_step}: {e}")
            ctx.failure_reason = str(e)
            await self._compensate(ctx)
        
        return ctx
    
    async def _create_order(self, ctx: SagaContext):
        order = await self.order_client.create(
            user_id=ctx.user_id,
            items=ctx.items,
            saga_id=ctx.saga_id,
        )
        ctx.order_id = order["id"]
        ctx.current_step = SagaStep.ORDER_CREATED
        ctx.completed_steps.append(SagaStep.ORDER_CREATED)
        await self._save_state(ctx)
    
    async def _reserve_inventory(self, ctx: SagaContext):
        await self.inventory_client.reserve(
            items=ctx.items,
            saga_id=ctx.saga_id,
        )
        ctx.current_step = SagaStep.INVENTORY_RESERVED
        ctx.completed_steps.append(SagaStep.INVENTORY_RESERVED)
        await self._save_state(ctx)
    
    async def _charge_payment(self, ctx: SagaContext):
        payment = await self.payment_client.charge(
            user_id=ctx.user_id,
            amount=ctx.total,
            saga_id=ctx.saga_id,
        )
        ctx.payment_id = payment["id"]
        ctx.current_step = SagaStep.PAYMENT_CHARGED
        ctx.completed_steps.append(SagaStep.PAYMENT_CHARGED)
        await self._save_state(ctx)
    
    async def _create_shipment(self, ctx: SagaContext):
        shipment = await self.shipping_client.create(
            order_id=ctx.order_id,
            saga_id=ctx.saga_id,
        )
        ctx.shipment_id = shipment["id"]
        ctx.current_step = SagaStep.SHIPMENT_CREATED
        ctx.completed_steps.append(SagaStep.SHIPMENT_CREATED)
        await self._save_state(ctx)
    
    async def _compensate(self, ctx: SagaContext):
        """اجرای compensation به ترتیب معکوس"""
        ctx.current_step = SagaStep.COMPENSATING
        await self._save_state(ctx)
        
        # ترتیب معکوس
        for step in reversed(ctx.completed_steps):
            try:
                if step == SagaStep.SHIPMENT_CREATED:
                    await self._cancel_shipment(ctx)
                elif step == SagaStep.PAYMENT_CHARGED:
                    await self._refund_payment(ctx)
                elif step == SagaStep.INVENTORY_RESERVED:
                    await self._release_inventory(ctx)
                elif step == SagaStep.ORDER_CREATED:
                    await self._cancel_order(ctx)
            except Exception as e:
                # Compensation حتی در صورت خطا باید ادامه یابد
                # و در صورت لزوم retry
                logger.error(f"Compensation {step} failed: {e}")
                await self._schedule_retry(ctx, step)
        
        ctx.current_step = SagaStep.FAILED
        await self._save_state(ctx)
    
    async def _release_inventory(self, ctx: SagaContext):
        await self.inventory_client.release(saga_id=ctx.saga_id)
        ctx.current_step = SagaStep.INVENTORY_RELEASED
        await self._save_state(ctx)
    
    async def _refund_payment(self, ctx: SagaContext):
        await self.payment_client.refund(payment_id=ctx.payment_id)
        ctx.current_step = SagaStep.PAYMENT_REFUNDED
        await self._save_state(ctx)
    
    async def _cancel_order(self, ctx: SagaContext):
        await self.order_client.cancel(
            order_id=ctx.order_id,
            reason=ctx.failure_reason,
        )
        ctx.current_step = SagaStep.ORDER_CANCELLED
        await self._save_state(ctx)
    
    async def _cancel_shipment(self, ctx: SagaContext):
        await self.shipping_client.cancel(shipment_id=ctx.shipment_id)
    
    async def _save_state(self, ctx: SagaContext):
        """ذخیره state در DB برای recovery"""
        await db.update_saga_state(
            saga_id=ctx.saga_id,
            state=asdict(ctx),
        )
    
    async def _schedule_retry(self, ctx, step):
        """retry compensation در صورت خطا"""
        await db.schedule_retry(saga_id=ctx.saga_id, step=step)

# استفاده
@app.post("/orders")
async def place_order(data: OrderCreate):
    ctx = SagaContext(
        saga_id=str(uuid.uuid4()),
        user_id=data.user_id,
        items=data.items,
        total=data.total,
    )
    
    orchestrator = OrderSagaOrchestrator()
    result = await orchestrator.execute(ctx)
    
    if result.current_step == SagaStep.COMPLETED:
        return {"status": "success", "order_id": result.order_id}
    else:
        return {"status": "failed", "reason": result.failure_reason}, 400
    

مزایا و معایب Orchestration

۷.۷ مقایسه: Choreography vs Orchestration

۷.۸ Framework های آماده Saga

Temporal

یکی از قدرتمندترین platform های workflow و saga. به‌صورت Open Source و SaaS موجود است.

# Temporal Python SDK
from temporalio import workflow, activity
from datetime import timedelta

@activity.defn
async def create_order(user_id: int, items: list) -> int:
    # call order service
    return order_id

@activity.defn
async def reserve_inventory(items: list) -> str:
    return reservation_id

@activity.defn
async def charge_payment(user_id: int, amount: float) -> int:
    return payment_id

@activity.defn
async def release_inventory(reservation_id: str):
    pass

@activity.defn
async def refund_payment(payment_id: int):
    pass

@workflow.defn
class OrderSagaWorkflow:
    @workflow.run
    async def run(self, user_id: int, items: list, total: float) -> dict:
        order_id = None
        reservation_id = None
        payment_id = None
        
        try:
            # Step 1
            order_id = await workflow.execute_activity(
                create_order,
                args=[user_id, items],
                start_to_close_timeout=timedelta(seconds=30),
            )
            
            # Step 2
            reservation_id = await workflow.execute_activity(
                reserve_inventory,
                args=[items],
                start_to_close_timeout=timedelta(seconds=30),
            )
            
            # Step 3
            payment_id = await workflow.execute_activity(
                charge_payment,
                args=[user_id, total],
                start_to_close_timeout=timedelta(seconds=30),
                retry_policy=workflow.RetryPolicy(
                    initial_interval=timedelta(seconds=1),
                    maximum_attempts=3,
                ),
            )
            
            return {"status": "success", "order_id": order_id}
            
        except Exception as e:
            # Compensation
            if payment_id:
                await workflow.execute_activity(
                    refund_payment, args=[payment_id]
                )
            if reservation_id:
                await workflow.execute_activity(
                    release_inventory, args=[reservation_id]
                )
            raise
    

سایر Framework ها

• AWS Step Functions: Managed، خوب برای AWS users
• Camunda: BPMN-based، قدرتمند برای enterprise
• Apache Airflow: برای data pipeline ها
• Cadence: precursor of Temporal از Uber
• Zeebe: از Camunda، cloud-native
• Restate.dev: جدید و promising

۷.۹ چالش‌های Saga

۱. Lack of Isolation

بین مراحل saga، داده‌ها در حالت intermediate هستند. این می‌تواند به «dirty reads» منجر شود.

راه‌حل‌ها:

• Semantic Lock: فلگ “in_progress” روی entity
• Pessimistic View: hide entity تا saga تمام شود
• Re-read: قبل از commit، دوباره بخوان
• Version Number: optimistic locking

۲. Idempotency

پیام‌ها ممکن است چند بار delivered شوند. هر مرحله باید idempotent باشد.

async def reserve_inventory(items, saga_id):
    # بررسی duplicate با saga_id
    existing = await db.get_reservation(saga_id)
    if existing:
        return existing  # idempotent
    
    # ایجاد رزرو جدید
    reservation = await db.create_reservation(saga_id, items)
    return reservation
    

۳. Compensation که Fail می‌شود

اگر compensation هم fail کند چه؟

راه‌حل: Compensation باید eventually successful باشد. با retry بی‌نهایت + manual intervention queue.

۴. Monitoring و Observability

تشخیص اینکه یک saga در کجا گیر کرده دشوار است.

راه‌حل:

• ذخیره state هر saga در DB
• Distributed tracing (Jaeger، Zipkin)
• Saga dashboard برای visibility
• Alert برای saga های stuck

۷.۱۰ بهترین تجربیات

1. هر مرحله را idempotent طراحی کنید. پیام ممکن است چند بار delivered شود.
2. State saga را persist کنید. برای recovery پس از crash.
3. Compensation همیشه باید موفق شود. با retry بی‌نهایت + alerting.
4. Timeout برای هر مرحله. از infinite waiting جلوگیری کنید.
5. Saga ID در همه پیام‌ها. برای tracking و correlation.
6. برای saga پیچیده، Orchestration. قابل debug تر است.
7. برای saga ساده، Choreography. سادگی و loose coupling.
8. Framework آماده استفاده کنید. Temporal، AWS Step Functions…
9. Monitoring قوی. dashboard برای saga های in-progress.
10. Test recovery. chaos engineering برای saga ها.
11. Saga نباید طولانی باشد. اگر ۱۰+ مرحله دارد، شاید طراحی اشتباه است.
12. Documentation. هر saga مستند شود (مراحل، compensation، error scenarios).

۷.۱۱ خلاصه فصل