Saga Pattern: Quản lý giao dịch phân tán trong kiến trúc Microservices

Saga Pattern là một mẫu thiết kế (design pattern) dùng để quản lý giao dịch giữa nhiều service độc lập trong kiến trúc microservices

Trong hệ thống monolithic, các giao dịch xảy ra giữa nhiều bảng dữ liệu có thể được quản lý bằng ACID transaction của cơ sở dữ liệu. Tuy nhiên, khi chuyển sang microservices:

Mỗi service có cơ sở dữ liệu riêng (Database per Service pattern)
Không thể thực hiện distributed ACID transactions qua nhiều cơ sở dữ liệu

Nếu một trong các service gặp lỗi, làm sao hệ thống có thể đảm bảo dữ liệu không rơi vào trạng thái không nhất quán?

Đây chính là lý do Saga Pattern được sử dụng - nó cung cấp một cách tiếp cận hiệu quả để quản lý các transaction phân tán, đảm bảo hệ thống vừa duy trì được tính nhất quán dữ liệu (eventual consistency), vừa không gây ra lỗi lan truyền khiến toàn bộ quy trình bị gián đoạn.

1. Saga Pattern là gì?

Saga Pattern là một mẫu thiết kế (design pattern) dùng để quản lý giao dịch giữa nhiều service độc lập trong kiến trúc microservices.

Trong hệ thống microservices, mỗi service thường có cơ sở dữ liệu riêng. Vì vậy, việc thực hiện một transaction xuyên suốt tất cả các service không thể sử dụng cách làm thông thường như với một cơ sở dữ liệu duy nhất (transaction ACID). Đây là lý do tại sao chúng ta cần Saga Pattern.

Hiểu một cách đơn giản, Saga là một chuỗi các transaction nhỏ và độc lập:

Mỗi transaction sẽ được thực hiện trong một service riêng lẻ, sau đó trigger (kích hoạt) transaction tiếp theo.
Nếu một transaction nào đó gặp lỗi, Saga sẽ thực hiện transaction hoàn tác (compensating transaction) để khôi phục lại dữ liệu về trạng thái ban đầu, đảm bảo hệ thống không bị sai lệch.

Saga Pattern giúp giữ cho dữ liệu trong hệ thống luôn nhất quán, ngay cả khi xảy ra lỗi ở một hoặc một số service.

2. Các phương pháp triển khai Saga Pattern

2.1 Choreography-based Saga

Trong mô hình này, các service hoạt động dựa trên cơ chế publish-subscribe events:

Mỗi service sẽ phát một sự kiện (event) khi hoàn thành công việc của mình.
Các service khác sẽ lắng nghe và phản ứng với sự kiện đó.
Không có một điểm điều phối trung tâm nào trong toàn bộ quy trình.

Ví dụ về luồng thành công:

Order Service: Tạo đơn hàng → Phát sự kiện ORDER_CREATED.
Payment Service: Lắng nghe sự kiện ORDER_CREATED → Xử lý thanh toán → Phát sự kiện PAYMENT_COMPLETED.
Inventory Service: Lắng nghe sự kiện PAYMENT_COMPLETED → Kiểm tra và giảm tồn kho → Phát sự kiện INVENTORY_UPDATED.
Delivery Service: Lắng nghe sự kiện INVENTORY_UPDATED → Lên lịch giao hàng → Phát sự kiện DELIVERY_SCHEDULED.
Order Service: Lắng nghe sự kiện DELIVERY_SCHEDULED → Cập nhật trạng thái đơn hàng thành CONFIRMED.

Ví dụ về luồng thất bại:

Order Service: Tạo đơn hàng → Phát sự kiện ORDER_CREATED.
Payment Service: Lắng nghe sự kiện ORDER_CREATED → Xử lý thanh toán → Phát sự kiện PAYMENT_COMPLETED.
Inventory Service: Lắng nghe sự kiện PAYMENT_COMPLETED → Kiểm tra tồn kho không đủ → Phát sự kiện INVENTORY_FAILED.
Payment Service: Lắng nghe sự kiện INVENTORY_FAILED → Hoàn tiền → Phát sự kiện PAYMENT_REFUNDED.
Order Service: Lắng nghe sự kiện PAYMENT_REFUNDED → Cập nhật trạng thái đơn hàng thành CANCELLED.

Điểm mạnh của mô hình này là các service độc lập với nhau, không phụ thuộc vào một điều phối trung tâm, dễ dàng mở rộng (scalable). Tuy nhiên điểm yếu của nó là khó kiểm soát luồng làm việc tổng thể, nhất là khi có nhiều service và sự kiện.

Javascript

// order-service/src/events/handlers.ts
import { Kafka, Consumer } from 'kafkajs';
import { OrderRepository } from '../repositories/order-repository';
import { OrderStatus } from '../models/order';

export class OrderEventHandlers {
  private consumer: Consumer;
  private orderRepository: OrderRepository;
  
  constructor(kafka: Kafka, orderRepository: OrderRepository) {
    this.consumer = kafka.consumer({ groupId: 'order-service' });
    this.orderRepository = orderRepository;
  }
  
  async start(): Promise<void> {
    await this.consumer.connect();
    await this.consumer.subscribe({ topic: 'payment-refunded', fromBeginning: true });
    
    await this.consumer.run({
      eachMessage: async ({ topic, partition, message }) => {
        if (topic === 'payment-refunded') {
          const event = JSON.parse(message.value!.toString());
          await this.handlePaymentRefunded(event);
        }
      },
    });
  }
  
  private async handlePaymentRefunded(event: { orderId: string }): Promise<void> {
    const order = await this.orderRepository.findById(event.orderId);
    if (!order) {
      console.error(`Order ${event.orderId} not found`);
      return;
    }
    
    order.status = OrderStatus.CANCELLED;
    await this.orderRepository.save(order);
    console.log(`Order ${event.orderId} cancelled due to inventory shortage`);
  }
  
  async stop(): Promise<void> {
    await this.consumer.disconnect();
  }
}

// payment-service/src/events/handlers.ts
import { Kafka, Consumer, Producer } from 'kafkajs';
import { PaymentRepository } from '../repositories/payment-repository';
import { RefundService } from '../services/refund-service';

export class PaymentEventHandlers {
  private consumer: Consumer;
  private producer: Producer;
  private paymentRepository: PaymentRepository;
  private refundService: RefundService;
  
  constructor(
    kafka: Kafka, 
    paymentRepository: PaymentRepository, 
    refundService: RefundService
  ) {
    this.consumer = kafka.consumer({ groupId: 'payment-service' });
    this.producer = kafka.producer();
    this.paymentRepository = paymentRepository;
    this.refundService = refundService;
  }
  
  async start(): Promise<void> {
    await this.consumer.connect();
    await this.producer.connect();
    await this.consumer.subscribe({ topic: 'inventory-failed', fromBeginning: true });
    
    await this.consumer.run({
      eachMessage: async ({ topic, partition, message }) => {
        if (topic === 'inventory-failed') {
          const event = JSON.parse(message.value!.toString());
          await this.handleInventoryFailed(event);
        }
      },
    });
  }
  
  private async handleInventoryFailed(event: { orderId: string }): Promise<void> {
    // Hoàn tiền
    const payment = await this.paymentRepository.findByOrderId(event.orderId);
    if (!payment) {
      console.error(`Payment for order ${event.orderId} not found`);
      return;
    }
    
    await this.refundService.refundPayment(payment);
    
    // Phát event
    await this.producer.send({
      topic: 'payment-refunded',
      messages: [
        { 
          key: event.orderId, 
          value: JSON.stringify({ orderId: event.orderId }) 
        },
      ],
    });
    
    console.log(`Payment refunded for order ${event.orderId}`);
  }
  
  async stop(): Promise<void> {
    await this.consumer.disconnect();
    await this.producer.disconnect();
  }
}

2.2 Orchestration-based Saga

Trong mô hình này, một central orchestrator sẽ chịu trách nhiệm điều phối toàn bộ quy trình giao tiếp giữa các service. Thay vì để các service tự phát và xử lý các sự kiện (như trong mô hình Choreography), orchestrator sẽ trực tiếp gọi các service phù hợp và quyết định bước tiếp theo trong quy trình.

Cách hoạt động:

Saga Orchestrator nhận yêu cầu tạo đơn hàng từ phía người dùng.
Orchestrator gọi Order Service để tạo đơn hàng với trạng thái ban đầu là PENDING.
Sau khi đơn hàng được tạo, Orchestrator gọi Payment Service để xử lý thanh toán.
Nếu thanh toán thành công, Orchestrator tiếp tục gọi Inventory Service để giảm tồn kho.
Nếu việc giảm tồn kho thành công, Orchestrator sẽ gọi Delivery Service để lên lịch giao hàng.
Cuối cùng, Orchestrator cập nhật trạng thái đơn hàng thành CONFIRMED.

Nếu bất kỳ bước nào trong các service thất bại (ví dụ: thanh toán không thành công hoặc tồn kho không đủ), Orchestrator sẽ thực hiện các bước bù trừ (compensating transactions) theo thứ tự ngược lại.

Ví dụ:

Nếu Inventory Service thất bại, Orchestrator sẽ gọi Payment Service để hoàn tiền.
Sau đó, Orchestrator sẽ cập nhật trạng thái đơn hàng từ PENDING thành CANCELLED.

Ưu điểm của Orchestration-based Saga là d kiểm soát toàn bộ luồng xử lý, vì mọi logic đều tập trung ởorchestrator. Debug và theo dõi lỗi dễ dàng hơn, vì orchestrator nắm rõ từng bước trong quy trình. Tuy nhiên cũng vì vậy mà orchestrator có thể trở thành điểm lỗi duy nhất (single point of failure) do mô hình này khiến Orchestrator phải gánh nhiều trách nhiệm, dễ dẫn đến quá tải nếu quy trình phức tạp.

Javascript

// order-saga/src/sagas/order-saga.ts
import { call, put, takeLatest } from 'redux-saga/effects';
import axios from 'axios';
import { 
  createOrderSuccess, 
  createOrderFailure,
  processPaymentSuccess,
  processPaymentFailure,
  // ... các action creators khác
} from '../actions/order-actions';

// Types
interface OrderRequest {
  customerId: string;
  items: Array<{
    productId: string;
    quantity: number;
    price: number;
  }>;
  totalAmount: number;
}

enum OrderState {
  STARTED = 'STARTED',
  ORDER_CREATED = 'ORDER_CREATED',
  PAYMENT_PROCESSED = 'PAYMENT_PROCESSED',
  INVENTORY_RESERVED = 'INVENTORY_RESERVED',
  DELIVERY_SCHEDULED = 'DELIVERY_SCHEDULED',
  COMPLETED = 'COMPLETED',
  CANCELLED = 'CANCELLED',
}

// Saga Workers
function* createOrder(action: { type: string, payload: OrderRequest }) {
  try {
    const { data } = yield call(
      axios.post, 
      'http://order-service/orders', 
      action.payload
    );
    yield put(createOrderSuccess(data));
    yield put({ type: 'PROCESS_PAYMENT', payload: { 
      orderId: data.id, 
      amount: action.payload.totalAmount 
    }});
  } catch (error) {
    yield put(createOrderFailure(error));
    // Không cần compensating transaction ở bước đầu tiên
  }
}

function* processPayment(action: { type: string, payload: { orderId: string, amount: number }}) {
  try {
    const { data } = yield call(
      axios.post, 
      'http://payment-service/payments', 
      {
        orderId: action.payload.orderId,
        amount: action.payload.amount,
      }
    );
    yield put(processPaymentSuccess(data));
    yield put({ type: 'RESERVE_INVENTORY', payload: { 
      orderId: action.payload.orderId,
      // Thêm dữ liệu cần thiết
    }});
  } catch (error) {
    yield put(processPaymentFailure(error));
    // Compensating transaction: hủy đơn hàng
    yield put({ type: 'CANCEL_ORDER', payload: { orderId: action.payload.orderId }});
  }
}

function* reserveInventory(action: { type: string, payload: { orderId: string }}) {
  try {
    const { data } = yield call(
      axios.post, 
      'http://inventory-service/inventory/reserve', 
      {
        orderId: action.payload.orderId,
        // Thêm dữ liệu cần thiết
      }
    );
    yield put({ type: 'RESERVE_INVENTORY_SUCCESS', payload: data });
    yield put({ type: 'SCHEDULE_DELIVERY', payload: { 
      orderId: action.payload.orderId,
      // Thêm dữ liệu cần thiết
    }});
  } catch (error) {
    yield put({ type: 'RESERVE_INVENTORY_FAILURE', payload: error });
    // Compensating transaction: hoàn tiền và hủy đơn hàng
    yield put({ type: 'REFUND_PAYMENT', payload: { orderId: action.payload.orderId }});
    yield put({ type: 'CANCEL_ORDER', payload: { orderId: action.payload.orderId }});
  }
}

// Compensating transaction workers
function* cancelOrder(action: { type: string, payload: { orderId: string }}) {
  try {
    yield call(
      axios.put, 
      `http://order-service/orders/${action.payload.orderId}/cancel`
    );
    yield put({ type: 'CANCEL_ORDER_SUCCESS', payload: action.payload });
  } catch (error) {
    yield put({ type: 'CANCEL_ORDER_FAILURE', payload: error });
    console.error('Failed to cancel order', error);
    // Có thể cần một cơ chế retry hoặc manual intervention
  }
}

function* refundPayment(action: { type: string, payload: { orderId: string }}) {
  try {
    yield call(
      axios.post, 
      'http://payment-service/payments/refund', 
      { orderId: action.payload.orderId }
    );
    yield put({ type: 'REFUND_PAYMENT_SUCCESS', payload: action.payload });
  } catch (error) {
    yield put({ type: 'REFUND_PAYMENT_FAILURE', payload: error });
    console.error('Failed to refund payment', error);
    // Cần một cơ chế retry hoặc manual intervention
  }
}

// Saga Watcher
export function* orderSaga() {
  yield takeLatest('CREATE_ORDER', createOrder);
  yield takeLatest('PROCESS_PAYMENT', processPayment);
  yield takeLatest('RESERVE_INVENTORY', reserveInventory);
  yield takeLatest('SCHEDULE_DELIVERY', scheduleDelivery);
  yield takeLatest('CANCEL_ORDER', cancelOrder);
  yield takeLatest('REFUND_PAYMENT', refundPayment);
  // ... các saga watchers khác
}

3. Kết luận

Saga Pattern là một giải pháp hiệu quả để quản lý các giao dịch phân tán trong kiến trúc microservices. Với hai cách triển khai chính - Choreography và Orchestration, pattern này mang đến sự linh hoạt trong việc xử lý các quy trình nghiệp vụ phức tạp, đồng thời đảm bảo dữ liệu luôn nhất quán.

Mặc dù phức tạp hơn so với các giao dịch ACID truyền thống, Saga Pattern lại giúp hệ thống phân tán hiện đại trở nên mạnh mẽ hơn nhờ khả năng mở rộng tốt, sẵn sàng cao, và chống chịu lỗi hiệu quả.

NovuTech