【系统架构设计百科】应用层数据一致性模式：在正确性与性能之间走钢丝

BEGIN;
UPDATE inventory SET stock = stock - 1 WHERE product_id = 1001;
UPDATE account SET balance = balance - 99.00 WHERE user_id = 42;
INSERT INTO orders (user_id, product_id, amount) VALUES (42, 1001, 99.00);
COMMIT;

// 库存服务的事件处理器
@EventHandler
public void on(OrderCreated event) {
    Inventory inv = inventoryRepository.findByProductId(event.getProductId());
    if (inv.getStock() < event.getQuantity()) {
        eventBus.publish(new InventoryInsufficient(event.getOrderId()));
        return;
    }
    inv.reserve(event.getQuantity());
    inventoryRepository.save(inv);
    eventBus.publish(new InventoryReserved(event.getOrderId(), event.getProductId()));
}

@EventHandler
public void on(PaymentFailed event) {
    // 补偿操作：释放已预留的库存
    Inventory inv = inventoryRepository.findByProductId(event.getProductId());
    inv.release(event.getQuantity());
    inventoryRepository.save(inv);
    eventBus.publish(new InventoryReleased(event.getOrderId()));
}

public class OrderSagaOrchestrator {

    private final OrderService orderService;
    private final InventoryService inventoryService;
    private final PaymentService paymentService;

    public enum SagaState {
        STARTED, INVENTORY_RESERVED, PAYMENT_PROCESSED,
        ORDER_CONFIRMED, COMPENSATING, FAILED
    }

    public void execute(CreateOrderCommand cmd) {
        SagaState state = SagaState.STARTED;
        String sagaId = UUID.randomUUID().toString();

        try {
            // 步骤 1：创建订单
            Order order = orderService.createOrder(cmd);
            sagaLog.record(sagaId, "ORDER_CREATED", order.getId());

            // 步骤 2：预留库存
            inventoryService.reserve(order.getProductId(), order.getQuantity());
            sagaLog.record(sagaId, "INVENTORY_RESERVED", order.getId());
            state = SagaState.INVENTORY_RESERVED;

            // 步骤 3：处理支付
            paymentService.processPayment(order.getUserId(), order.getAmount());
            sagaLog.record(sagaId, "PAYMENT_PROCESSED", order.getId());
            state = SagaState.PAYMENT_PROCESSED;

            // 步骤 4：确认订单
            orderService.confirmOrder(order.getId());
            sagaLog.record(sagaId, "ORDER_CONFIRMED", order.getId());
            state = SagaState.ORDER_CONFIRMED;

        } catch (InventoryInsufficientException e) {
            compensate(sagaId, state, cmd);
        } catch (PaymentFailedException e) {
            compensate(sagaId, state, cmd);
        }
    }

    private void compensate(String sagaId, SagaState failedAt, CreateOrderCommand cmd) {
        sagaLog.record(sagaId, "COMPENSATING", failedAt.name());

        switch (failedAt) {
            case PAYMENT_PROCESSED:
                // 不应该到这里，支付成功后不会触发补偿
                break;
            case INVENTORY_RESERVED:
                // 支付失败，回滚库存
                inventoryService.release(cmd.getProductId(), cmd.getQuantity());
                sagaLog.record(sagaId, "INVENTORY_RELEASED", "");
                // fall through
            case STARTED:
                // 取消订单
                orderService.cancelOrder(cmd.getOrderId());
                sagaLog.record(sagaId, "ORDER_CANCELLED", "");
                break;
        }
    }
}

public class InventoryService {

    public void release(String reservationId) {
        // 通过 reservationId 保证幂等
        Reservation reservation = reservationRepository.findById(reservationId);
        if (reservation == null || reservation.getStatus() == ReservationStatus.RELEASED) {
            // 已释放或不存在，直接返回
            return;
        }
        Inventory inv = inventoryRepository.findByProductId(reservation.getProductId());
        inv.setStock(inv.getStock() + reservation.getQuantity());
        reservation.setStatus(ReservationStatus.RELEASED);

        // 在同一个本地事务中更新库存和预留记录
        inventoryRepository.save(inv);
        reservationRepository.save(reservation);
    }
}

public void cancelPayment(String paymentId) {
    Payment payment = paymentRepository.findById(paymentId);
    if (payment == null) {
        // 空补偿：支付请求从未到达，直接成功
        return;
    }
    if (payment.getStatus() == PaymentStatus.CANCELLED) {
        // 幂等：已取消，直接成功
        return;
    }
    payment.setStatus(PaymentStatus.CANCELLED);
    accountService.unfreeze(payment.getUserId(), payment.getAmount());
    paymentRepository.save(payment);
}

@Scheduled(fixedDelay = 5000)
public void retryFailedCompensations() {
    List<CompensationTask> pending = compensationTaskRepository
        .findByStatusAndRetryCountLessThan(
            CompensationStatus.PENDING, MAX_RETRIES);

    for (CompensationTask task : pending) {
        try {
            executeCompensation(task);
            task.setStatus(CompensationStatus.COMPLETED);
        } catch (Exception e) {
            task.setRetryCount(task.getRetryCount() + 1);
            task.setNextRetryAt(calculateBackoff(task.getRetryCount()));
            if (task.getRetryCount() >= MAX_RETRIES) {
                task.setStatus(CompensationStatus.FAILED);
                alertService.notify("Compensation failed permanently: " + task.getId());
            }
        }
        compensationTaskRepository.save(task);
    }
}

-- 预留阶段：不直接修改 stock，而是增加 reserved
UPDATE inventory
SET reserved = reserved + 1
WHERE product_id = 1001 AND (stock - reserved) >= 1;

-- 确认阶段：Saga 全部成功后
UPDATE inventory
SET stock = stock - 1, reserved = reserved - 1
WHERE product_id = 1001;

-- 补偿阶段：Saga 失败后
UPDATE inventory
SET reserved = reserved - 1
WHERE product_id = 1001;

// 库存服务的 TCC 接口
public interface InventoryTccService {

    /**
     * Try：预留库存。
     * 冻结指定数量的库存，但不真正扣减。
     * 幂等：同一个 xid 重复调用返回成功。
     */
    boolean tryReserve(String xid, String productId, int quantity);

    /**
     * Confirm：确认扣减。
     * 将 Try 阶段冻结的库存真正扣减。
     * 幂等：同一个 xid 重复调用返回成功。
     */
    boolean confirm(String xid);

    /**
     * Cancel：释放预留。
     * 解冻 Try 阶段冻结的库存。
     * 幂等：同一个 xid 重复调用返回成功。
     */
    boolean cancel(String xid);
}

// 库存服务的 TCC 实现
public class InventoryTccServiceImpl implements InventoryTccService {

    @Override
    @Transactional
    public boolean tryReserve(String xid, String productId, int quantity) {
        // 幂等检查
        TccRecord record = tccRecordRepository.findByXid(xid);
        if (record != null) {
            return record.getStatus() != TccStatus.CANCELLED;
        }

        Inventory inv = inventoryRepository.findByProductId(productId);
        if (inv.getAvailable() < quantity) {
            return false;
        }

        // 冻结库存：available 减少，frozen 增加
        inv.setAvailable(inv.getAvailable() - quantity);
        inv.setFrozen(inv.getFrozen() + quantity);
        inventoryRepository.save(inv);

        // 记录 TCC 事务
        tccRecordRepository.save(new TccRecord(xid, productId, quantity, TccStatus.TRIED));
        return true;
    }

    @Override
    @Transactional
    public boolean confirm(String xid) {
        TccRecord record = tccRecordRepository.findByXid(xid);
        if (record == null) {
            return false;
        }
        if (record.getStatus() == TccStatus.CONFIRMED) {
            return true; // 幂等
        }

        Inventory inv = inventoryRepository.findByProductId(record.getProductId());
        // 真正扣减：frozen 减少
        inv.setFrozen(inv.getFrozen() - record.getQuantity());
        inventoryRepository.save(inv);

        record.setStatus(TccStatus.CONFIRMED);
        tccRecordRepository.save(record);
        return true;
    }

    @Override
    @Transactional
    public boolean cancel(String xid) {
        TccRecord record = tccRecordRepository.findByXid(xid);
        if (record == null) {
            return true; // 空取消，直接成功
        }
        if (record.getStatus() == TccStatus.CANCELLED) {
            return true; // 幂等
        }

        Inventory inv = inventoryRepository.findByProductId(record.getProductId());
        // 解冻：frozen 减少，available 恢复
        inv.setFrozen(inv.getFrozen() - record.getQuantity());
        inv.setAvailable(inv.getAvailable() + record.getQuantity());
        inventoryRepository.save(inv);

        record.setStatus(TccStatus.CANCELLED);
        tccRecordRepository.save(record);
        return true;
    }
}

type TccCoordinator struct {
    inventorySvc InventoryTccService
    paymentSvc   PaymentTccService
    orderSvc     OrderTccService
}

func (c *TccCoordinator) ProcessOrder(ctx context.Context, cmd CreateOrderCmd) error {
    xid := uuid.New().String()

    // Phase 1: Try
    if err := c.orderSvc.TryCreate(ctx, xid, cmd); err != nil {
        return fmt.Errorf("order try failed: %w", err)
    }
    if err := c.inventorySvc.TryReserve(ctx, xid, cmd.ProductID, cmd.Quantity); err != nil {
        c.orderSvc.Cancel(ctx, xid)
        return fmt.Errorf("inventory try failed: %w", err)
    }
    if err := c.paymentSvc.TryFreeze(ctx, xid, cmd.UserID, cmd.Amount); err != nil {
        c.inventorySvc.Cancel(ctx, xid)
        c.orderSvc.Cancel(ctx, xid)
        return fmt.Errorf("payment try failed: %w", err)
    }

    // Phase 2: Confirm
    // 所有 Try 都成功后，执行 Confirm。
    // Confirm 失败需要重试（通常由框架保证）。
    if err := c.orderSvc.Confirm(ctx, xid); err != nil {
        log.Printf("order confirm failed, will retry: %v", err)
    }
    if err := c.inventorySvc.Confirm(ctx, xid); err != nil {
        log.Printf("inventory confirm failed, will retry: %v", err)
    }
    if err := c.paymentSvc.Confirm(ctx, xid); err != nil {
        log.Printf("payment confirm failed, will retry: %v", err)
    }

    return nil
}

// 定时任务：释放超时的 TCC 预留
@Scheduled(fixedDelay = 10000)
public void releaseExpiredReservations() {
    LocalDateTime cutoff = LocalDateTime.now().minusMinutes(5);
    List<TccRecord> expired = tccRecordRepository
        .findByStatusAndCreatedAtBefore(TccStatus.TRIED, cutoff);
    for (TccRecord record : expired) {
        cancel(record.getXid());
        log.warn("Released expired TCC reservation: {}", record.getXid());
    }
}

BEGIN;
-- 业务操作
UPDATE account SET balance = balance - 100 WHERE user_id = 42;

-- 写入消息表（同一个数据库）
INSERT INTO outbox_messages (id, topic, payload, status, created_at)
VALUES ('msg-001', 'payment.completed',
        '{"orderId": "ord-123", "userId": 42, "amount": 100}',
        'PENDING', NOW());

COMMIT;

@Component
public class OutboxMessageRelay {

    private final OutboxMessageRepository messageRepository;
    private final MessageBroker messageBroker;

    @Scheduled(fixedDelay = 1000)
    @Transactional
    public void publishPendingMessages() {
        List<OutboxMessage> pending = messageRepository
            .findByStatusOrderByCreatedAtAsc(MessageStatus.PENDING, PageRequest.of(0, 100));

        for (OutboxMessage msg : pending) {
            try {
                messageBroker.publish(msg.getTopic(), msg.getPayload());
                msg.setStatus(MessageStatus.SENT);
                msg.setSentAt(LocalDateTime.now());
            } catch (Exception e) {
                msg.setRetryCount(msg.getRetryCount() + 1);
                if (msg.getRetryCount() > MAX_RETRIES) {
                    msg.setStatus(MessageStatus.FAILED);
                    log.error("Outbox message permanently failed: {}", msg.getId(), e);
                }
            }
            messageRepository.save(msg);
        }
    }
}

@MessageHandler
public void handlePaymentCompleted(PaymentCompletedEvent event) {
    // 幂等检查：使用消息 ID 去重
    if (processedMessageRepository.existsById(event.getMessageId())) {
        log.info("Message already processed: {}", event.getMessageId());
        return;
    }

    // 在同一个事务中处理业务逻辑和记录消息 ID
    orderService.confirmPayment(event.getOrderId());
    processedMessageRepository.save(
        new ProcessedMessage(event.getMessageId(), LocalDateTime.now()));
}

CREATE TABLE outbox (
    id            BIGINT PRIMARY KEY AUTO_INCREMENT,
    aggregate_type VARCHAR(255) NOT NULL,
    aggregate_id   VARCHAR(255) NOT NULL,
    event_type     VARCHAR(255) NOT NULL,
    payload        JSON NOT NULL,
    created_at     TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);

{
    "name": "outbox-connector",
    "config": {
        "connector.class": "io.debezium.connector.mysql.MySqlConnector",
        "database.hostname": "order-db",
        "database.port": "3306",
        "database.user": "debezium",
        "database.password": "******",
        "database.server.id": "1",
        "database.server.name": "order-service",
        "table.include.list": "orders.outbox",
        "transforms": "outbox",
        "transforms.outbox.type":
            "io.debezium.transforms.outbox.EventRouter",
        "transforms.outbox.table.field.event.id": "id",
        "transforms.outbox.table.field.event.key": "aggregate_id",
        "transforms.outbox.table.field.event.type": "event_type",
        "transforms.outbox.table.field.event.payload": "payload",
        "transforms.outbox.route.by.field": "aggregate_type",
        "transforms.outbox.route.topic.replacement":
            "outbox.event.${routedByValue}"
    }
}

-- 清理 24 小时前的已处理消息
DELETE FROM outbox WHERE created_at < DATE_SUB(NOW(), INTERVAL 24 HOUR) LIMIT 10000;

// Eventuate Tram 的 MessageProducer 内部实现（简化版）
public class MessageProducerImpl implements MessageProducer {

    private final MessageRepository messageRepository;

    @Override
    @Transactional
    public void send(String destination, Message message) {
        message.setId(idGenerator.genId());
        message.setDestination(destination);
        message.setHeaders(enrichHeaders(message.getHeaders()));
        // 关键：消息写入本地数据库，与业务操作在同一个事务中
        messageRepository.save(toMessageEntity(message));
    }
}

CREATE TABLE message (
    id            VARCHAR(767) PRIMARY KEY,
    destination   VARCHAR(1000) NOT NULL,
    headers       VARCHAR(1000) NOT NULL,
    payload       TEXT NOT NULL,
    published     SMALLINT DEFAULT 0,
    creation_time BIGINT
);

public class CreateOrderSaga implements SimpleSaga<CreateOrderSagaData> {

    private final SagaDefinition<CreateOrderSagaData> sagaDefinition;

    public CreateOrderSaga(OrderServiceProxy orderService,
                           InventoryServiceProxy inventoryService,
                           PaymentServiceProxy paymentService) {
        this.sagaDefinition =
            step()
                .invokeLocal(this::createOrder)
                .withCompensation(this::rejectOrder)
            .step()
                .invokeParticipant(inventoryService.reserve)
                .withCompensation(inventoryService.release)
            .step()
                .invokeParticipant(paymentService.process)
                .withCompensation(paymentService.refund)
            .step()
                .invokeLocal(this::approveOrder)
            .build();
    }

    @Override
    public SagaDefinition<CreateOrderSagaData> getSagaDefinition() {
        return sagaDefinition;
    }

    private void createOrder(CreateOrderSagaData data) {
        // 创建订单，状态为 PENDING
        Order order = orderRepository.save(
            new Order(data.getUserId(), data.getProductId(),
                      data.getQuantity(), OrderStatus.PENDING));
        data.setOrderId(order.getId());
    }

    private void rejectOrder(CreateOrderSagaData data) {
        orderRepository.updateStatus(data.getOrderId(), OrderStatus.REJECTED);
    }

    private void approveOrder(CreateOrderSagaData data) {
        orderRepository.updateStatus(data.getOrderId(), OrderStatus.APPROVED);
    }
}

CREATE TABLE saga_instance (
    saga_type        VARCHAR(255) NOT NULL,
    saga_id          VARCHAR(100) NOT NULL,
    state_name       VARCHAR(255) NOT NULL,
    last_request_id  VARCHAR(100),
    end_state        TINYINT(1),
    compensating     TINYINT(1),
    saga_data_type   VARCHAR(1000) NOT NULL,
    saga_data_json   TEXT NOT NULL,
    PRIMARY KEY (saga_type, saga_id)
);

public class InventoryCommandHandler {

    @CommandHandler
    public Message reserve(CommandMessage<ReserveInventoryCommand> cm) {
        ReserveInventoryCommand cmd = cm.getCommand();
        try {
            inventoryService.reserve(cmd.getProductId(), cmd.getQuantity());
            return withSuccess(new InventoryReserved(cmd.getProductId()));
        } catch (InsufficientInventoryException e) {
            return withFailure(new InventoryInsufficient(cmd.getProductId()));
        }
    }

    @CommandHandler
    public Message release(CommandMessage<ReleaseInventoryCommand> cm) {
        ReleaseInventoryCommand cmd = cm.getCommand();
        inventoryService.release(cmd.getProductId(), cmd.getQuantity());
        return withSuccess(new InventoryReleased(cmd.getProductId()));
    }
}

public class CreateOrderSaga implements SimpleSaga<CreateOrderSagaData> {

    private final SagaDefinition<CreateOrderSagaData> sagaDefinition;

    public CreateOrderSaga() {
        this.sagaDefinition =
            step()
                .invokeLocal(this::createPendingOrder)
                .withCompensation(this::rejectOrder)
            .step()
                .invokeParticipant(this::reserveInventory)
                .onReply(InventoryReserved.class, this::onInventoryReserved)
                .withCompensation(this::releaseInventory)
            .step()
                .invokeParticipant(this::requestPayment)
                .onReply(PaymentCompleted.class, this::onPaymentCompleted)
                .withCompensation(this::refundPayment)
            .step()
                .invokeLocal(this::approveOrder)
            .step()
                .invokeParticipant(this::sendNotification)
                // 通知不需要补偿——发了就发了
            .build();
    }

    private void createPendingOrder(CreateOrderSagaData data) {
        Order order = new Order();
        order.setUserId(data.getUserId());
        order.setProductId(data.getProductId());
        order.setQuantity(data.getQuantity());
        order.setAmount(data.getAmount());
        order.setStatus(OrderStatus.PENDING);
        order = orderRepository.save(order);
        data.setOrderId(order.getId());
    }

    private CommandWithDestination reserveInventory(CreateOrderSagaData data) {
        return send(new ReserveInventoryCommand(
                data.getOrderId(), data.getProductId(), data.getQuantity()))
            .to("inventoryService")
            .build();
    }

    private CommandWithDestination releaseInventory(CreateOrderSagaData data) {
        return send(new ReleaseInventoryCommand(
                data.getOrderId(), data.getProductId(), data.getQuantity()))
            .to("inventoryService")
            .build();
    }

    private CommandWithDestination requestPayment(CreateOrderSagaData data) {
        return send(new ProcessPaymentCommand(
                data.getOrderId(), data.getUserId(), data.getAmount()))
            .to("paymentService")
            .build();
    }

    private CommandWithDestination refundPayment(CreateOrderSagaData data) {
        return send(new RefundPaymentCommand(
                data.getOrderId(), data.getPaymentId()))
            .to("paymentService")
            .build();
    }

    private void approveOrder(CreateOrderSagaData data) {
        orderRepository.updateStatus(data.getOrderId(), OrderStatus.APPROVED);
    }

    private void rejectOrder(CreateOrderSagaData data) {
        orderRepository.updateStatus(data.getOrderId(), OrderStatus.REJECTED);
    }

    private void onInventoryReserved(CreateOrderSagaData data,
                                      InventoryReserved reply) {
        data.setReservationId(reply.getReservationId());
    }

    private void onPaymentCompleted(CreateOrderSagaData data,
                                     PaymentCompleted reply) {
        data.setPaymentId(reply.getPaymentId());
    }

    private CommandWithDestination sendNotification(CreateOrderSagaData data) {
        return send(new SendOrderConfirmationCommand(
                data.getOrderId(), data.getUserId()))
            .to("notificationService")
            .build();
    }
}

public class PaymentCommandHandler {

    @CommandHandler
    public Message refund(CommandMessage<RefundPaymentCommand> cm) {
        RefundPaymentCommand cmd = cm.getCommand();
        Payment payment = paymentRepository.findById(cmd.getPaymentId());

        if (payment == null || payment.getStatus() == PaymentStatus.REFUNDED) {
            return withSuccess(new PaymentRefunded(cmd.getPaymentId()));
        }

        // 发起异步退款请求
        String refundId = paymentGateway.initiateRefund(payment.getTransactionId(),
                                                         payment.getAmount());
        payment.setStatus(PaymentStatus.REFUND_PENDING);
        payment.setRefundId(refundId);
        paymentRepository.save(payment);

        // 立即返回成功——退款结果由回调处理
        return withSuccess(new PaymentRefundInitiated(cmd.getPaymentId(), refundId));
    }

    // 第三方支付网关的退款回调
    @PostMapping("/webhook/refund")
    public void handleRefundCallback(@RequestBody RefundCallbackRequest req) {
        Payment payment = paymentRepository.findByRefundId(req.getRefundId());
        if ("SUCCESS".equals(req.getStatus())) {
            payment.setStatus(PaymentStatus.REFUNDED);
        } else {
            payment.setStatus(PaymentStatus.REFUND_FAILED);
            alertService.notify("Refund failed: " + req.getRefundId());
        }
        paymentRepository.save(payment);
    }
}

CREATE TABLE compensation_log (
    id              BIGINT PRIMARY KEY AUTO_INCREMENT,
    saga_id         VARCHAR(100) NOT NULL,
    step_name       VARCHAR(100) NOT NULL,
    action          VARCHAR(20) NOT NULL,  -- 'FORWARD' or 'COMPENSATE'
    status          VARCHAR(20) NOT NULL,  -- 'SUCCESS', 'FAILED', 'PENDING'
    request_payload TEXT,
    response_payload TEXT,
    retry_count     INT DEFAULT 0,
    created_at      TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at      TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    INDEX idx_saga_id (saga_id),
    INDEX idx_status (status)
);

// Go 语言的 Saga 监控指标
var (
    sagaStarted = prometheus.NewCounter(prometheus.CounterOpts{
        Name: "saga_started_total",
        Help: "Total number of sagas started",
    })
    sagaCompleted = prometheus.NewCounter(prometheus.CounterOpts{
        Name: "saga_completed_total",
        Help: "Total number of sagas completed successfully",
    })
    sagaCompensated = prometheus.NewCounter(prometheus.CounterOpts{
        Name: "saga_compensated_total",
        Help: "Total number of sagas that triggered compensation",
    })
    sagaFailed = prometheus.NewCounter(prometheus.CounterOpts{
        Name: "saga_failed_total",
        Help: "Total number of sagas that failed permanently",
    })
    sagaDuration = prometheus.NewHistogram(prometheus.HistogramOpts{
        Name:    "saga_duration_seconds",
        Help:    "Duration of saga execution",
        Buckets: prometheus.ExponentialBuckets(0.01, 2, 15),
    })
    compensationRetries = prometheus.NewHistogram(prometheus.HistogramOpts{
        Name:    "compensation_retries",
        Help:    "Number of retries before compensation succeeds",
        Buckets: prometheus.LinearBuckets(0, 1, 10),
    })
)

// 管理后台接口：手动触发 Saga 补偿或强制完成
@RestController
@RequestMapping("/admin/sagas")
public class SagaAdminController {

    @PostMapping("/{sagaId}/retry-compensation")
    public ResponseEntity<?> retryCompensation(@PathVariable String sagaId) {
        SagaInstance saga = sagaRepository.findById(sagaId);
        if (saga == null) {
            return ResponseEntity.notFound().build();
        }
        if (!saga.isCompensating()) {
            return ResponseEntity.badRequest()
                .body("Saga is not in compensating state");
        }
        sagaEngine.retryCompensation(sagaId);
        return ResponseEntity.ok("Compensation retry initiated");
    }

    @PostMapping("/{sagaId}/force-complete")
    public ResponseEntity<?> forceComplete(@PathVariable String sagaId) {
        // 强制标记 Saga 为完成状态
        // 仅在人工确认数据已手动修复后使用
        SagaInstance saga = sagaRepository.findById(sagaId);
        saga.setEndState(true);
        saga.setCompensating(false);
        sagaRepository.save(saga);
        auditLog.record("SAGA_FORCE_COMPLETE", sagaId,
            SecurityContextHolder.getContext().getAuthentication().getName());
        return ResponseEntity.ok("Saga force completed");
    }
}

@Test
void sagaShouldCompensateWhenPaymentFails() {
    // Given
    when(inventoryService.reserve(any(), anyInt()))
        .thenReturn(new InventoryReserved("res-1"));
    when(paymentService.process(any(), any(), any()))
        .thenThrow(new PaymentFailedException("Insufficient balance"));

    // When
    sagaOrchestrator.execute(new CreateOrderCommand("user-1", "prod-1", 1, 99.00));

    // Then
    // 验证补偿操作按逆序执行
    InOrder inOrder = inOrder(inventoryService, orderService);
    inOrder.verify(inventoryService).release("prod-1", 1);
    inOrder.verify(orderService).cancelOrder(anyString());

    // 验证订单状态
    Order order = orderRepository.findByUserId("user-1");
    assertEquals(OrderStatus.REJECTED, order.getStatus());
}

@Test
void compensationShouldBeIdempotent() {
    // Given：模拟补偿被执行两次
    inventoryService.reserve("res-1", "prod-1", 1);

    // When：第一次补偿
    inventoryService.release("res-1");
    int stockAfterFirst = inventoryRepository.findByProductId("prod-1").getStock();

    // When：第二次补偿（重复）
    inventoryService.release("res-1");
    int stockAfterSecond = inventoryRepository.findByProductId("prod-1").getStock();

    // Then：库存只释放了一次
    assertEquals(stockAfterFirst, stockAfterSecond);
}

@Transactional
public void handleCompensation(String operationId) {
    OperationRecord record = operationRepository
        .findByIdForUpdate(operationId); // SELECT ... FOR UPDATE
    if (record == null) {
        // 正向操作尚未开始，记录一个"预补偿"标记
        operationRepository.save(new OperationRecord(operationId,
            OperationStatus.PRE_COMPENSATED));
        return;
    }
    if (record.getStatus() == OperationStatus.IN_PROGRESS) {
        throw new ConcurrentOperationException(
            "Operation in progress, retry later");
    }
    // 正向操作已完成，执行补偿
    doCompensate(record);
}

-- 使用分区表，按天分区
CREATE TABLE outbox (
    id            BIGINT NOT NULL AUTO_INCREMENT,
    aggregate_type VARCHAR(255) NOT NULL,
    aggregate_id   VARCHAR(255) NOT NULL,
    event_type     VARCHAR(255) NOT NULL,
    payload        JSON NOT NULL,
    created_at     TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    PRIMARY KEY (id, created_at)
) PARTITION BY RANGE (UNIX_TIMESTAMP(created_at)) (
    PARTITION p20260413 VALUES LESS THAN (UNIX_TIMESTAMP('2026-04-14')),
    PARTITION p20260414 VALUES LESS THAN (UNIX_TIMESTAMP('2026-04-15')),
    PARTITION pmax VALUES LESS THAN MAXVALUE
);

-- 清理：直接 DROP 旧分区，比 DELETE 快几个数量级
ALTER TABLE outbox DROP PARTITION p20260411;