Mục lục nội dung

1. Vấn đề Dual-Write trong Kiến trúc Phân tán và Hậu quả

Trong các ứng dụng backend hiện đại xây dựng theo kiến trúc Microservices hoặc Event-Driven Architecture, một thao tác nghiệp vụ thường đòi hỏi hệ thống phải làm đồng thời hai việc: cập nhật dữ liệu vào cơ sở dữ liệu quan hệ (RDBMS) và phát một sự kiện (Event) tới Message Broker (như RabbitMQ, Apache Kafka, hoặc NATS). Thao tác này được gọi là Dual-Write (Ghi kép).

Hãy xét ví dụ về một dịch vụ đặt hàng (Order Service). Khi người dùng tạo đơn hàng, hệ thống cần thực hiện:

  1. Lưu thông tin đơn hàng mới vào cơ sở dữ liệu PostgreSQL thông qua TypeORM với trạng thái CREATED.
  2. Phát sự kiện OrderCreatedEvent sang RabbitMQ để Notification Service gửi email và Inventory Service trừ kho.

Trong môi trường phân tán, hai thao tác này không thể gộp chung vào một ACID Transaction mang tính chất nguyên tử (Atomic). Điều gì sẽ xảy ra nếu một trong hai thao tác thất bại?

  • Trường hợp 1: Ghi Database thành công, nhưng ứng dụng bị crash hoặc mất kết nối mạng ngay trước khi gửi message tới RabbitMQ. Kết quả: Đơn hàng đã được lưu, nhưng email không được gửi và hàng trong kho không bị trừ. Dữ liệu bị bất đồng bộ.
  • Trường hợp 2: Gửi message tới RabbitMQ trước, sau đó lưu Database nhưng thất bại do lỗi Constraint Violation. Kết quả: Notification Service nhận sự kiện và gửi email thông báo thành công cho khách hàng, nhưng đơn hàng thực tế lại không tồn tại trong hệ thống.

Các giải pháp truyền thống như Two-Phase Commit (2PC) hay Distributed Transactions tỏ ra quá nặng nề, giảm đáng kể throughput và dễ gây ra deadlock trên diện rộng. Đây chính là lúc Transactional Outbox Pattern trở thành chuẩn mực thiết kế hàng đầu.

2. Bản chất Kỹ thuật của Transactional Outbox Pattern

Nguyên lý cốt lõi của Transactional Outbox Pattern rất đơn giản nhưng triệt để: Thay vì trực tiếp gửi sự kiện sang Message Broker trong HTTP Request lifecycle, chúng ta ghi sự kiện đó vào một bảng tạm nằm trong cùng một Database Transaction với dữ liệu nghiệp vụ.

Bảng tạm này được gọi là Outbox Table. Vì cả thao tác lưu nghiệp vụ và lưu bản tin Outbox đều diễn ra trên cùng một kết nối Database, tính chất ACID đảm bảo rằng: Hoặc là cả hai đều thành công, hoặc cả hai đều bị rollback.

Sau khi transaction commit thành công, một tiến trình chạy ngầm (Background Worker hoặc Outbox Relay) sẽ đảm nhiệm việc đọc các sự kiện chưa xử lý từ Outbox Table, phát chúng lên Message Broker và cập nhật lại trạng thái bản tin là đã gửi.

3. Thiết kế và Triển khai với NestJS và TypeORM

Dưới đây là mô hình triển khai chi tiết từng bước Transactional Outbox Pattern trong một ứng dụng NestJS sử dụng TypeORM và PostgreSQL.

3.1. Định nghĩa Outbox Entity

Đầu tiên, chúng ta xây dựng Entity đại diện cho bảng Outbox trong CSDL.

import { Entity, PrimaryGeneratedColumn, Column, CreateDateColumn, Index } from 'typeorm';

export enum OutboxStatus {
  PENDING = 'PENDING',
  PROCESSED = 'PROCESSED',
  FAILED = 'FAILED',
}

@Entity('outbox_messages')
export class OutboxEntity {
  @PrimaryGeneratedColumn('uuid')
  id: string;

  @Column({ type: 'varchar', length: 255 })
  aggregateType: string;

  @Column({ type: 'varchar', length: 255 })
  aggregateId: string;

  @Column({ type: 'varchar', length: 255 })
  eventType: string;

  @Column({ type: 'jsonb' })
  payload: Record<string, any>;

  @Index()
  @Column({
    type: 'enum',
    enum: OutboxStatus,
    default: OutboxStatus.PENDING,
  })
  status: OutboxStatus;

  @CreateDateColumn()
  createdAt: Date;

  @Column({ type: 'timestamp', nullable: true })
  processedAt: Date;

  @Column({ type: 'text', nullable: true })
  errorMessage: string;
}

3.2. Quản lý Atomic Transaction trong Service

Khi xử lý request tạo đơn hàng, chúng ta sử dụng QueryRunner của TypeORM để khởi tạo một transaction duy nhất bọc cả OrderEntity và OutboxEntity.

import { Injectable, InternalServerErrorException } from '@nestjs/common';
import { DataSource } from 'typeorm';
import { OrderEntity } from './entities/order.entity';
import { OutboxEntity, OutboxStatus } from './entities/outbox.entity';
import { CreateOrderDto } from './dto/create-order.dto';

@Injectable()
export class OrderService {
  constructor(private readonly dataSource: DataSource) {}

  async createOrder(createOrderDto: CreateOrderDto): Promise<OrderEntity> {
    const queryRunner = this.dataSource.createQueryRunner();

    await queryRunner.connect();
    await queryRunner.startTransaction();

    try {
      // 1. Tạo và lưu Order Entity
      const order = queryRunner.manager.create(OrderEntity, {
        customerId: createOrderDto.customerId,
        totalAmount: createOrderDto.totalAmount,
        status: 'PENDING',
      });
      const savedOrder = await queryRunner.manager.save(order);

      // 2. Tạo bản tin Outbox trong cùng Transaction
      const outboxEvent = queryRunner.manager.create(OutboxEntity, {
        aggregateType: 'ORDER',
        aggregateId: savedOrder.id,
        eventType: 'ORDER_CREATED',
        payload: {
          orderId: savedOrder.id,
          customerId: savedOrder.customerId,
          totalAmount: savedOrder.totalAmount,
          createdAt: savedOrder.createdAt,
        },
        status: OutboxStatus.PENDING,
      });
      await queryRunner.manager.save(outboxEvent);

      // 3. Commit toàn bộ thao tác
      await queryRunner.commitTransaction();
      return savedOrder;
    } catch (error) {
      // Rollback nếu có bất kỳ lỗi nào xuất hiện
      await queryRunner.rollbackTransaction();
      throw new InternalServerErrorException('Không thể khởi tạo đơn hàng: ' + error.message);
    } finally {
      await queryRunner.release();
    }
  }
}

4. Xây dựng Outbox Worker với Giải pháp Chống Concurrency Lock

Một thách thức lớn khi triển khai Polling Outbox trong môi trường Production là: Khi ứng dụng NestJS được scale ra nhiều Pods/Instances, nhiều worker sẽ cùng chạy một lúc và dẫn tới nguy cơ đọc trùng sự kiện (Race Condition).

Để giải quyết bài toán này, chúng ta kết hợp tính năng SKIP LOCKED của PostgreSQL trong câu lệnh SQL query của TypeORM. Tính năng này cho phép một instance khóa các hàng đang xử lý và cho phép các instance khác bỏ qua các hàng đó để xử lý các hàng tiếp theo mà không bị block thread.

import { Injectable, Logger } from '@nestjs/common';
import { Cron, CronExpression } from '@nestjs/schedule';
import { DataSource } from 'typeorm';
import { OutboxEntity, OutboxStatus } from './entities/outbox.entity';
import { MessageBrokerService } from '../broker/message-broker.service';

@Injectable()
export class OutboxProcessorWorker {
  private readonly logger = new Logger(OutboxProcessorWorker.name);

  constructor(
    private readonly dataSource: DataSource,
    private readonly messageBroker: MessageBrokerService,
  ) {}

  @Cron(CronExpression.EVERY_5_SECONDS)
  async processOutbox(): Promise<void> {
    const queryRunner = this.dataSource.createQueryRunner();
    await queryRunner.connect();
    await queryRunner.startTransaction();

    try {
      // Sử dụng pessimistic_write và setOnLocked('skip_locked') để ngăn race condition giữa nhiều Pods
      const pendingMessages = await queryRunner.manager
        .createQueryBuilder(OutboxEntity, 'outbox')
        .where('outbox.status = :status', { status: OutboxStatus.PENDING })
        .orderBy('outbox.createdAt', 'ASC')
        .take(20)
        .setLock('pessimistic_write')
        .setOnLocked('skip_locked')
        .getMany();

      if (pendingMessages.length === 0) {
        await queryRunner.commitTransaction();
        return;
      }

      for (const message of pendingMessages) {
        try {
          // Gửi message tới Message Broker (RabbitMQ / Kafka)
          await this.messageBroker.publish(
            message.eventType,
            message.payload,
          );

          message.status = OutboxStatus.PROCESSED;
          message.processedAt = new Date();
          await queryRunner.manager.save(message);
        } catch (publishError) {
          this.logger.error(`Lỗi gửi message ${message.id}: ${publishError.message}`);
          message.status = OutboxStatus.FAILED;
          message.errorMessage = publishError.message;
          await queryRunner.manager.save(message);
        }
      }

      await queryRunner.commitTransaction();
    } catch (error) {
      await queryRunner.rollbackTransaction();
      this.logger.error(`Lỗi hệ thống trong Outbox Worker: ${error.message}`);
    } finally {
      await queryRunner.release();
    }
  }
}

5. Đảm bảo Nguyên tắc At-Least-Once và Idempotent Consumer

Transactional Outbox Pattern cung cấp cơ chế phân phối tin nhắn theo chuẩn At-Least-Once Delivery (Đảm bảo tin nhắn được gửi ít nhất một lần). Điều này có nghĩa là trong một số trường hợp rủi ro (ví dụ: Worker đã phát tin thành công tới RabbitMQ nhưng chưa kịp update trạng thái bản tin trong DB thành PROCESSED thì bị sập server), tin nhắn đó có thể bị gửi lại ở lượt poll tiếp theo.

Do đó, phía dịch vụ tiêu thụ (Consumer) BẮT BUỘC phải triển khai cơ chế Idempotent Consumer (Xử lý trùng lặp dữ liệu).

Kỹ thuật triển khai Idempotency bằng Unique Consumer Log:

Consumer duy trì một bảng ghi nhận lịch sử các Message ID đã được xử lý thành công. Trước khi thực thi logic nghiệp vụ, Consumer kiểm tra Message ID đã tồn tại chưa.

import { Injectable } from '@nestjs/common';
import { DataSource } from 'typeorm';
import { ProcessedEventEntity } from './entities/processed-event.entity';

@Injectable()
export class InventoryConsumerService {
  constructor(private readonly dataSource: DataSource) {}

  async handleOrderCreated(eventPayload: any, messageId: string): Promise<void> {
    const queryRunner = this.dataSource.createQueryRunner();
    await queryRunner.connect();
    await queryRunner.startTransaction();

    try {
      // 1. Kiểm tra xem messageId đã được xử lý thành công trước đó chưa
      const isAlreadyProcessed = await queryRunner.manager.findOne(ProcessedEventEntity, {
        where: { messageId },
      });

      if (isAlreadyProcessed) {
        // Nếu đã xử lý rồi, bỏ qua không làm gì cả để đảm bảo Idempotency
        await queryRunner.rollbackTransaction();
        return;
      }

      // 2. Thực thi logic nghiệp vụ (Ví dụ: Trừ kho)
      await this.deductInventory(queryRunner.manager, eventPayload);

      // 3. Đánh dấu message ID đã xử lý
      const processedRecord = queryRunner.manager.create(ProcessedEventEntity, {
        messageId,
        processedAt: new Date(),
      });
      await queryRunner.manager.save(processedRecord);

      await queryRunner.commitTransaction();
    } catch (error) {
      await queryRunner.rollbackTransaction();
      throw error;
    } finally {
      await queryRunner.release();
    }
  }

  private async deductInventory(manager: any, payload: any): Promise<void> {
    // Logic cập nhật kho hàng ở đây
  }
}

6. So sánh Polling Outbox vs Change Data Capture (CDC)

Có hai cách tiếp cận chính để đọc dữ liệu từ bảng Outbox gửi lên Broker:

Têu chí Polling Publisher (Phương pháp trên) Change Data Capture (CDC - Debezium)
Độ phức tạp hạ tầng Thấp, chỉ cần viết code Worker trong NestJS. Cao, cần triển khai Kafka Connect, Debezium, ZooKeeper/KRaft.
Tác động lên Database Tạo tải truy vấn (SELECT polling) lên RDBMS theo chu kỳ. Gần như bằng 0, đọc trực tiếp từ Write-Ahead Log (WAL) của PostgreSQL.
Latency (Độ trễ) Phụ thuộc vào Cron Job interval (1s - 5s). Near Real-time (< 100ms).
Phù hợp cho dự án Vừa và nhỏ, microservices có tải vừa phải. Hệ thống Enterprise quy mô cực lớn, High-throughput.

7. Kết luận

Transactional Outbox Pattern là chìa khóa vàng giúp hệ thống backend phân tán giải quyết triệt để bài toán Dual-Write, đảm bảo tính nhất quán dữ liệu (Data Consistency) giữa Database và Message Broker mà không làm suy giảm hiệu năng của ứng dụng.

Việc làm chủ các pattern nâng cao cùng kỹ năng quản lý Database Transaction chính xác với TypeORM và kiến trúc khung NestJS là yếu tố cốt lõi của một Backend Engineer chuyên nghiệp. Tham khảo khóa học "RESTful API với NestJS & TypeORM" tại đây để làm chủ tư duy thiết kế hệ thống và xây dựng các ứng dụng chuẩn Enterprise.