Saga Pattern

1
2
3
4
5
6

[Diagram: Saga Pattern Flow]
┌─────────────┐   ┌─────────────┐   ┌─────────────┐   ┌─────────────┐
│   Service A │ → │   Service B │ → │   Service C │ → │   Service D │
└─────────────┘   └─────────────┘   └─────────────┘   └─────────────┘
      |                 |                 |                 |
      └────Compensate───┴────Compensate───┴────Compensate───┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

// 간단한 오케스트레이션 기반 Saga 예시
class SagaOrchestrator {
  constructor(services) {
    this.services = services;
    this.sagaLog = [];
  }

  async executeSaga() {
    try {
      for (const service of this.services) {
        await service.execute();
        this.sagaLog.push({ service: service.name, status: 'success' });
      }
      console.log('Saga completed successfully');
    } catch (error) {
      console.error('Saga failed, compensating...');
      for (let i = this.sagaLog.length - 1; i >= 0; i--) {
        await this.services[i].compensate();
        this.sagaLog[i].status = 'compensated';
      }
      console.log('Saga compensated');
    }
  }
}

// 서비스 예시
const orderService = {
  name: 'Order',
  execute: async () => { console.log('Order created'); },
  compensate: async () => { console.log('Order cancelled'); }
};
const paymentService = {
  name: 'Payment',
  execute: async () => { console.log('Payment processed'); },
  compensate: async () => { console.log('Payment refunded'); }
};

const orchestrator = new SagaOrchestrator([orderService, paymentService]);
orchestrator.executeSaga();

1

Distributed-Transactions, Compensating-Transactions, Orchestration-vs-Choreography, Event-Driven

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

@Service
public class OrderSagaOrchestrator {
  @Autowired private KafkaTemplate<String, OrderEvent> kafka;
  @Autowired private SagaLogRepository sagaLog;

  @Transactional
  public void startSaga(CreateOrderCmd cmd) {
    sagaLog.save(new Saga(cmd.getOrderId(), "STARTED", ...));
    kafka.send("order", new OrderCreated(cmd.getOrderId(), ...));
  }

  @KafkaListener(topics="payments")
  public void onPaymentResult(PaymentResult evt) {
    Saga saga = sagaLog.find(evt.getOrderId());
    if (evt.isSuccess()) {
      kafka.send("stock", new ReserveStock(evt.getOrderId(), evt.getItems()));
      saga.updateStep("STOCK");
    } else {
      kafka.send("order", new CompensateOrder(evt.getOrderId()));
      saga.updateStatus("COMPENSATED");
    }
    sagaLog.save(saga);
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

# saga.py
class SagaStep:
    def __init__(self, action, compensation):
        self.action, self.compensation = action, compensation

class SagaOrchestrator:
    def __init__(self, steps):
        self.steps, self.completed = steps, []

    def run(self):
        for step in self.steps:
            try:
                step.action()
                self.completed.append(step)
            except Exception:
                for prev in reversed(self.completed):
                    prev.compensation()
                raise

# application example
def reserve_stock(): print('stock reserved')

def charge_payment(): print('payment charged')
def refund_payment(): print('payment refunded')

steps = [
    SagaStep(reserve_stock, lambda: print('stock released')),
    SagaStep(charge_payment, refund_payment)
]

SagaOrchestrator(steps).run()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

from dataclasses import dataclass

@dataclass
class OrderCreated:
    order_id: str
    items: list

@dataclass
class PaymentSucceeded:
    order_id: str
    amount: float

@dataclass
class PaymentFailed:
    order_id: str
    reason: str

@dataclass
class StockReleased:
    order_id: str

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

import pika, json

class Broker:
    def __init__(self):
        self.conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
        self.ch = self.conn.channel()
        self.ch.exchange_declare(exchange='saga', exchange_type='topic')

    def publish(self, routing_key, event):
        self.ch.basic_publish(
            exchange='saga',
            routing_key=routing_key,
            body=json.dumps(event.__dict__)
        )

    def subscribe(self, routing_key, callback):
        q = self.ch.queue_declare('', exclusive=True).method.queue
        self.ch.queue_bind(exchange='saga', queue=q, routing_key=routing_key)
        self.ch.basic_consume(q, lambda ch, method, prop, body: callback(json.loads(body)), auto_ack=True)
        self.ch.start_consuming()

1
2
3
4
5
6
7
8

from events import OrderCreated
from messaging import Broker
import uuid

broker = Broker()
order_id = str(uuid.uuid4())
broker.publish('order.created', OrderCreated(order_id, ['item1', 'item2']))
print(f"OrderCreated published for {order_id}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

from events import OrderCreated, PaymentSucceeded, PaymentFailed
from messaging import Broker

broker = Broker()

def on_order_created(data):
    order_id = data['order_id']
    # 결제 시뮬레이션
    if True:  # 결제 성공 조건
        broker.publish('payment.succeeded', PaymentSucceeded(order_id, 100.0))
    else:
        broker.publish('payment.failed', PaymentFailed(order_id, "Insufficient funds"))

broker.subscribe('order.created', on_order_created)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

from events import PaymentSucceeded, PaymentFailed, StockReleased
from messaging import Broker

broker = Broker()

def on_payment_succeeded(data):
    order_id = data['order_id']
    # 재고 차감 시뮬레이션
    print(f"Stock reserved for {order_id}")

def on_payment_failed(data):
    order_id = data['order_id']
    broker.publish('stock.released', StockReleased(order_id))

broker.subscribe('payment.succeeded', on_payment_succeeded)
broker.subscribe('payment.failed', on_payment_failed)
broker.subscribe('stock.released', lambda data: print(f"Stock released for {data['order_id']}"))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# tracing.py
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.jaeger.thrift import JaegerExporter

trace.set_tracer_provider(TracerProvider())
jaeger_exporter = JaegerExporter(agent_host_name="localhost", agent_port=6831)
trace.get_tracer_provider().add_span_processor(BatchSpanProcessor(jaeger_exporter))
tracer = trace.get_tracer(__name__)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# service_a.py
from tracing import tracer
from messaging import Broker
from events import OrderCreated

broker = Broker()

def create_order(...):
    with tracer.start_as_current_span("create_order_saga"):
        order_id = ...
        broker.publish("order.created", OrderCreated(order_id, ...))

1
2
3
4
5
6
7

# service_b.py
from tracing import tracer

def on_order_created(msg):
    with tracer.start_as_current_span("payment_step") as span:
        span.set_attribute("order_id", msg["order_id"])
        # 결제 시도…

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

# 주문 서비스
class OrderService:
    def create_order(self, order_data):
        order = Order.create(order_data)
        self.event_bus.publish(OrderCreatedEvent(order.id))
        return order

# 결제 서비스
class PaymentService:
    @event_handler
    def handle_order_created(self, event):
        payment_result = self.process_payment(event.order_id)
        if payment_result.success:
            self.event_bus.publish(PaymentProcessedEvent(event.order_id))
        else:
            self.event_bus.publish(PaymentFailedEvent(event.order_id))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

class TravelBookingSaga:
    def __init__(self):
        self.state = "STARTED"
        self.bookings = {}
    
    def execute(self, booking_request):
        try:
            # 항공편 예약
            flight_booking = self.flight_service.book(booking_request.flight)
            self.bookings['flight'] = flight_booking
            
            # 호텔 예약
            hotel_booking = self.hotel_service.book(booking_request.hotel)
            self.bookings['hotel'] = hotel_booking
            
            # 렌터카 예약
            car_booking = self.car_service.book(booking_request.car)
            self.bookings['car'] = car_booking
            
            self.state = "COMPLETED"
            return True
            
        except Exception as e:
            self.compensate()
            return False
    
    def compensate(self):
        if 'car' in self.bookings:
            self.car_service.cancel(self.bookings['car'].id)
        if 'hotel' in self.bookings:
            self.hotel_service.cancel(self.bookings['hotel'].id)
        if 'flight' in self.bookings:
            self.flight_service.cancel(self.bookings['flight'].id)
        self.state = "COMPENSATED"

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215

import asyncio
from enum import Enum
from dataclasses import dataclass
from typing import List, Dict, Optional
import uuid

class SagaStatus(Enum):
    STARTED = "STARTED"
    COMPLETED = "COMPLETED"
    COMPENSATING = "COMPENSATING"
    FAILED = "FAILED"

@dataclass
class OrderData:
    customer_id: str
    product_id: str
    quantity: int
    amount: float

@dataclass
class SagaStep:
    step_id: str
    service_name: str
    action: str
    compensating_action: str
    status: str = "PENDING"
    result: Dict = None

class OrderProcessingSaga:
    def __init__(self):
        self.saga_id = str(uuid.uuid4())
        self.status = SagaStatus.STARTED
        self.steps: List[SagaStep] = []
        self.completed_steps: List[str] = []
    
    async def execute_order_saga(self, order_data: OrderData):
        """주문 처리 사가 실행"""
        try:
            # Step 1: 주문 생성
            order_step = SagaStep(
                step_id="1",
                service_name="OrderService",
                action="create_order",
                compensating_action="cancel_order"
            )
            order_result = await self.create_order(order_data)
            order_step.result = order_result
            order_step.status = "COMPLETED"
            self.steps.append(order_step)
            self.completed_steps.append("1")
            
            # Step 2: 결제 처리
            payment_step = SagaStep(
                step_id="2",
                service_name="PaymentService",
                action="process_payment",
                compensating_action="refund_payment"
            )
            payment_result = await self.process_payment(
                order_result['order_id'], 
                order_data.amount
            )
            payment_step.result = payment_result
            payment_step.status = "COMPLETED"
            self.steps.append(payment_step)
            self.completed_steps.append("2")
            
            # Step 3: 재고 차감
            inventory_step = SagaStep(
                step_id="3",
                service_name="InventoryService",
                action="reserve_inventory",
                compensating_action="release_inventory"
            )
            inventory_result = await self.reserve_inventory(
                order_data.product_id,
                order_data.quantity
            )
            inventory_step.result = inventory_result
            inventory_step.status = "COMPLETED"
            self.steps.append(inventory_step)
            self.completed_steps.append("3")
            
            # Step 4: 배송 준비
            shipping_step = SagaStep(
                step_id="4",
                service_name="ShippingService",
                action="prepare_shipment",
                compensating_action="cancel_shipment"
            )
            shipping_result = await self.prepare_shipment(
                order_result['order_id']
            )
            shipping_step.result = shipping_result
            shipping_step.status = "COMPLETED"
            self.steps.append(shipping_step)
            self.completed_steps.append("4")
            
            self.status = SagaStatus.COMPLETED
            return {"status": "SUCCESS", "saga_id": self.saga_id}
            
        except Exception as e:
            print(f"사가 실행 중 오류 발생: {e}")
            await self.compensate()
            return {"status": "FAILED", "saga_id": self.saga_id, "error": str(e)}
    
    async def compensate(self):
        """보상 트랜잭션 실행"""
        self.status = SagaStatus.COMPENSATING
        
        # 완료된 단계를 역순으로 보상
        for step_id in reversed(self.completed_steps):
            step = next(s for s in self.steps if s.step_id == step_id)
            
            try:
                if step.service_name == "ShippingService":
                    await self.cancel_shipment(step.result['shipment_id'])
                elif step.service_name == "InventoryService":
                    await self.release_inventory(
                        step.result['product_id'],
                        step.result['quantity']
                    )
                elif step.service_name == "PaymentService":
                    await self.refund_payment(step.result['payment_id'])
                elif step.service_name == "OrderService":
                    await self.cancel_order(step.result['order_id'])
                    
                step.status = "COMPENSATED"
                print(f"Step {step_id} 보상 완료")
                
            except Exception as e:
                print(f"보상 트랜잭션 실패 - Step {step_id}: {e}")
                step.status = "COMPENSATION_FAILED"
        
        self.status = SagaStatus.FAILED
    
    # 각 서비스 메서드들 (실제로는 HTTP 호출)
    async def create_order(self, order_data: OrderData):
        """주문 생성 서비스 호출"""
        # 시뮬레이션: 주문 서비스 API 호출
        await asyncio.sleep(0.1)
        order_id = str(uuid.uuid4())
        print(f"주문 생성 완료: {order_id}")
        return {"order_id": order_id, "status": "PENDING"}
    
    async def process_payment(self, order_id: str, amount: float):
        """결제 처리 서비스 호출"""
        await asyncio.sleep(0.1)
        # 10% 확률로 결제 실패 시뮬레이션
        if asyncio.get_event_loop().time() % 10 < 1:
            raise Exception("결제 처리 실패")
        
        payment_id = str(uuid.uuid4())
        print(f"결제 처리 완료: {payment_id}")
        return {"payment_id": payment_id, "amount": amount}
    
    async def reserve_inventory(self, product_id: str, quantity: int):
        """재고 예약 서비스 호출"""
        await asyncio.sleep(0.1)
        print(f"재고 예약 완료: {product_id}, 수량: {quantity}")
        return {"product_id": product_id, "quantity": quantity}
    
    async def prepare_shipment(self, order_id: str):
        """배송 준비 서비스 호출"""
        await asyncio.sleep(0.1)
        shipment_id = str(uuid.uuid4())
        print(f"배송 준비 완료: {shipment_id}")
        return {"shipment_id": shipment_id, "order_id": order_id}
    
    # 보상 트랜잭션 메서드들
    async def cancel_order(self, order_id: str):
        await asyncio.sleep(0.1)
        print(f"주문 취소 완료: {order_id}")
    
    async def refund_payment(self, payment_id: str):
        await asyncio.sleep(0.1)
        print(f"결제 환불 완료: {payment_id}")
    
    async def release_inventory(self, product_id: str, quantity: int):
        await asyncio.sleep(0.1)
        print(f"재고 해제 완료: {product_id}, 수량: {quantity}")
    
    async def cancel_shipment(self, shipment_id: str):
        await asyncio.sleep(0.1)
        print(f"배송 취소 완료: {shipment_id}")

# 사용 예시
async def main():
    """사가 패턴 실행 예시"""
    saga = OrderProcessingSaga()
    
    order_data = OrderData(
        customer_id="customer_123",
        product_id="product_456",
        quantity=2,
        amount=99.99
    )
    
    print("=== 주문 처리 사가 시작 ===")
    result = await saga.execute_order_saga(order_data)
    
    print(f"\n=== 사가 실행 결과 ===")
    print(f"상태: {result['status']}")
    print(f"사가 ID: {result['saga_id']}")
    
    if result['status'] == 'FAILED':
        print(f"오류: {result.get('error', 'Unknown error')}")
    
    print(f"\n=== 사가 단계 상세 ===")
    for step in saga.steps:
        print(f"Step {step.step_id} ({step.service_name}): {step.status}")

# 실행
if __name__ == "__main__":
    asyncio.run(main())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

public interface Saga {
    // Saga의 각 단계를 정의
    void execute();
    // 실패 시 보상 트랜잭션 실행
    void compensate();
}

// 주문 처리를 위한 Saga 구현
public class OrderSaga implements Saga {
    private final OrderService orderService;
    private final PaymentService paymentService;
    private final InventoryService inventoryService;
    private final List<SagaStep> completedSteps = new ArrayList<>();

    @Override
    public void execute() {
        try {
            // 1단계: 주문 생성
            Order order = orderService.createOrder();
            completedSteps.add(new OrderCreationStep(order));

            // 2단계: 결제 처리
            Payment payment = paymentService.processPayment(order);
            completedSteps.add(new PaymentProcessingStep(payment));

            // 3단계: 재고 감소
            inventoryService.decreaseStock(order);
            completedSteps.add(new StockDecrementStep(order));

        } catch (Exception e) {
            // 오류 발생 시 보상 트랜잭션 실행
            compensate();
            throw e;
        }
    }

    @Override
    public void compensate() {
        // 실행된 단계들을 역순으로 보상 처리
        for (int i = completedSteps.size() - 1; i >= 0; i--) {
            completedSteps.get(i).compensate();
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

public interface SagaStep {
    void execute();
    void compensate();
}

public class OrderCreationStep implements SagaStep {
    private final Order order;
    private final OrderService orderService;

    @Override
    public void execute() {
        orderService.createOrder(order);
    }

    @Override
    public void compensate() {
        orderService.cancelOrder(order.getId());
    }
}

public class PaymentStep implements SagaStep {
    private final Payment payment;
    private final PaymentService paymentService;

    @Override
    public void execute() {
        paymentService.processPayment(payment);
    }

    @Override
    public void compensate() {
        paymentService.refundPayment(payment.getId());
    }
}