fix(lastleg): harden container plan saga recovery

2026-09-09 00:31:50 +08:00
parent 990b74f3e8
commit 2ba5f0c307
8 changed files with 327 additions and 60 deletions
@@ -19,6 +19,7 @@ public class ContainerPlanConversionCommand {
    private Long warehouseId;
    private String batchKey;
    private String status;
+    private String leaseToken;
    private String preparationJson;
    private String resultJson;
    private String lastError;
@@ -17,17 +17,33 @@ public interface ContainerPlanConversionCommandMapper extends BaseMapper<Contain
    @Select("SELECT * FROM ll_container_plan_conversion_command WHERE source_container_booking_id=#{bookingId} LIMIT 1")
    ContainerPlanConversionCommand findByBooking(@Param("bookingId") String bookingId);

-    @Update("UPDATE ll_container_plan_conversion_command SET status=#{status},result_json=COALESCE(#{resultJson},result_json),last_error=#{lastError},updated_at=#{updatedAt} WHERE id=#{id} AND status=#{expectedStatus}")
+    @Update("UPDATE ll_container_plan_conversion_command SET status=#{status},result_json=COALESCE(#{resultJson},result_json),last_error=#{lastError},updated_at=#{updatedAt} WHERE id=#{id} AND status=#{expectedStatus} AND lease_token=#{leaseToken}")
    int updateState(@Param("id") long id, @Param("expectedStatus") String expectedStatus,
+                    @Param("leaseToken") String leaseToken,
                    @Param("status") String status,
                    @Param("resultJson") String resultJson, @Param("lastError") String lastError,
                    @Param("updatedAt") LocalDateTime updatedAt);

-    @Update("UPDATE ll_container_plan_conversion_command SET status=#{nextStatus},last_error=NULL,updated_at=#{updatedAt} WHERE id=#{id} AND status=#{expectedStatus}")
+    @Update("UPDATE ll_container_plan_conversion_command SET status=#{nextStatus},lease_token=#{leaseToken},last_error=NULL,updated_at=#{updatedAt} WHERE id=#{id} AND status=#{expectedStatus}")
    int claim(@Param("id") long id, @Param("expectedStatus") String expectedStatus,
-              @Param("nextStatus") String nextStatus, @Param("updatedAt") LocalDateTime updatedAt);
+              @Param("nextStatus") String nextStatus, @Param("leaseToken") String leaseToken,
+              @Param("updatedAt") LocalDateTime updatedAt);

-    @Update("UPDATE ll_container_plan_conversion_command SET last_error=NULL,updated_at=#{updatedAt} WHERE id=#{id} AND status='APPLYING' AND updated_at=#{expectedUpdatedAt}")
+    @Update("UPDATE ll_container_plan_conversion_command SET lease_token=#{leaseToken},last_error=NULL,updated_at=#{updatedAt} WHERE id=#{id} AND status='APPLYING' AND updated_at=#{expectedUpdatedAt}")
    int reclaimStale(@Param("id") long id, @Param("expectedUpdatedAt") LocalDateTime expectedUpdatedAt,
-                     @Param("updatedAt") LocalDateTime updatedAt);
+                     @Param("leaseToken") String leaseToken, @Param("updatedAt") LocalDateTime updatedAt);
+
+    @Update("""
+        UPDATE ll_container_plan_conversion_command
+           SET idempotency_key=#{key},request_hash=#{requestHash},snapshot_version=#{snapshotVersion},
+               snapshot_hash=#{snapshotHash},warehouse_id=#{warehouseId},batch_key=#{batchKey},
+               status='PREPARED',lease_token=NULL,preparation_json=#{preparationJson},result_json=NULL,last_error=NULL,
+               updated_at=#{updatedAt}
+         WHERE id=#{id} AND status='FAILED'
+        """)
+    int replaceFailed(@Param("id") long id, @Param("key") String key,
+                      @Param("requestHash") String requestHash, @Param("snapshotVersion") long snapshotVersion,
+                      @Param("snapshotHash") String snapshotHash, @Param("warehouseId") long warehouseId,
+                      @Param("batchKey") String batchKey, @Param("preparationJson") String preparationJson,
+                      @Param("updatedAt") LocalDateTime updatedAt);
 }
@@ -46,82 +46,108 @@ public class ContainerInboundPlanConversionCoordinator {
        String bookingId = required(request.sourceContainerBookingId(), 128, "来源柜预订 ID 不能为空");
        String key = required(idempotencyKey, 128, "Idempotency-Key 必须为1至128个字符");
        String requestHash = hash(List.of(bookingId, request.snapshotVersion(), request.snapshotHash()));
-        ContainerPlanConversionCommand resolved = commands.find(bookingId, key);
-        if (resolved == null) resolved = commands.findByBooking(bookingId);
-        if (resolved == null) resolved = createCommand(bookingId, key, request, requestHash);
+        ContainerPlanConversionCommand direct = commands.find(bookingId, key);
+        ContainerPlanConversionCommand resolved = direct == null ? commands.findByBooking(bookingId) : direct;
+        if (resolved == null) {
+            resolved = createCommand(bookingId, key, request, requestHash);
+        } else if ("COMPLETED".equals(resolved.getStatus())) {
+            if (direct != null) requireSameRequest(resolved, requestHash);
+            return result(resolved);
+        } else if ("FAILED".equals(resolved.getStatus())) {
+            resolved = replaceFailedCommand(resolved, bookingId, key, request, requestHash);
+        } else {
+            requireSameRequest(resolved, requestHash);
+        }
+        if ("COMPLETED".equals(resolved.getStatus())) {
+            requireSameRequest(resolved, requestHash);
+            return result(resolved);
+        }
        final ContainerPlanConversionCommand command = resolved;
-        requireSameRequest(command, requestHash);
        Preparation preparation = read(command.getPreparationJson(), Preparation.class);

-        if ("COMPLETED".equals(command.getStatus())) return result(command);
+        boolean applyingLeaseOwned = false;
        if ("APPLYING".equals(command.getStatus())) {
            LocalDateTime now = LocalDateTime.now();
+            String leaseToken = IdWorker.getIdStr();
            if (command.getUpdatedAt() == null || command.getUpdatedAt().plus(APPLYING_LEASE).isAfter(now)
-                || commands.reclaimStale(command.getId(), command.getUpdatedAt(), now) != 1) {
+                || commands.reclaimStale(command.getId(), command.getUpdatedAt(), leaseToken, now) != 1) {
                throw new ResponseStatusException(HttpStatus.CONFLICT, "该柜正在生成入库计划，请勿重复提交");
            }
+            command.setLeaseToken(leaseToken);
            command.setUpdatedAt(now);
+            applyingLeaseOwned = true;
        }
-        if (!"LOCAL_COMMITTED".equals(command.getStatus())) {
-            if (commands.claim(command.getId(), "PREPARED", "APPLYING", LocalDateTime.now()) != 1) {
+        if ("PREPARED".equals(command.getStatus())) {
+            String leaseToken = IdWorker.getIdStr();
+            if (commands.claim(command.getId(), "PREPARED", "APPLYING", leaseToken, LocalDateTime.now()) != 1) {
                ContainerPlanConversionCommand current = commands.findByBooking(bookingId);
                if (current != null && "COMPLETED".equals(current.getStatus())) return result(current);
                throw new ResponseStatusException(HttpStatus.CONFLICT, "该柜正在生成入库计划，请勿重复提交");
            }
            command.setStatus("APPLYING");
+            command.setLeaseToken(leaseToken);
+            applyingLeaseOwned = true;
+        }
+        if (!applyingLeaseOwned && !"LOCAL_COMMITTED".equals(command.getStatus())) {
+            throw new ResponseStatusException(HttpStatus.CONFLICT, "柜级转换命令状态无效");
        }
        if (preparation.items().isEmpty()) {
            try {
                List<InboundPlanResponse> existing = inTransaction(() -> {
                    List<InboundPlanResponse> converted = plans.convertContainerReserved(bookingId,
                        request.snapshotVersion(), request.snapshotHash(), List.of());
-                    state(command.getId(), "APPLYING", "COMPLETED", write(converted), null);
+                    state(command, "APPLYING", "COMPLETED", write(converted), null);
                    return converted;
                });
                command.setStatus("COMPLETED");
                return existing;
            } catch (RuntimeException failure) {
-                resetForRetry(command, failure);
+                ContainerPlanConversionCommand current = commands.findByBooking(bookingId);
+                if (current != null && "COMPLETED".equals(current.getStatus())) return result(current);
+                failWithoutResources(command, failure);
                throw failure;
            }
        }

-        WarehouseResourceGateway.PalletBatchEnsure ensure = new WarehouseResourceGateway.PalletBatchEnsure(
-            preparation.warehouseId(), command.getBatchKey(), preparation.items().stream().map(item ->
-                new WarehouseResourceGateway.PalletBatchItemRequest(Long.parseLong(item.customerId()), item.businessKey(),
-                    item.planNo(), item.globalSequenceNo(), "STANDARD", item.fbaWarehouseCode())).toList());
-        WarehouseResourceGateway.PalletBatchResponse reservation;
-        try {
-            reservation = warehouse.ensurePalletBatch(ensure);
-            requireBatch(reservation, command.getBatchKey(), preparation.items().size(), "RESERVED", "CONFIRMED");
-        } catch (RuntimeException failure) {
-            resetForRetry(command, failure);
-            throw failure;
-        }
-
        List<InboundPlanResponse> localResult;
        if ("LOCAL_COMMITTED".equals(command.getStatus())) {
            localResult = result(command);
        } else {
+            WarehouseResourceGateway.PalletBatchEnsure ensure = new WarehouseResourceGateway.PalletBatchEnsure(
+                preparation.warehouseId(), command.getBatchKey(), preparation.items().stream().map(item ->
+                    new WarehouseResourceGateway.PalletBatchItemRequest(Long.parseLong(item.customerId()),
+                        item.businessKey(), item.planNo(), item.globalSequenceNo(), "STANDARD",
+                        item.fbaWarehouseCode())).toList());
+            WarehouseResourceGateway.PalletBatchResponse reservation;
+            try {
+                reservation = warehouse.ensurePalletBatch(ensure);
+                requireBatch(reservation, command.getBatchKey(), preparation.items().size(), "RESERVED", "CONFIRMED");
+            } catch (RuntimeException failure) {
+                compensateAndFail(command, preparation, failure);
+                throw failure;
+            }
            try {
                List<InboundPlanService.PreparedPlan> preparedPlans = bind(preparation, reservation);
                localResult = inTransaction(() -> {
                    List<InboundPlanResponse> created = plans.convertContainerReserved(bookingId,
                        request.snapshotVersion(), request.snapshotHash(), preparedPlans);
-                    state(command.getId(), "APPLYING", "LOCAL_COMMITTED", write(created), null);
+                    state(command, "APPLYING", "LOCAL_COMMITTED", write(created), null);
                    return created;
                });
                command.setStatus("LOCAL_COMMITTED");
            } catch (RuntimeException failure) {
-                try {
-                    warehouse.compensatePalletBatch(new WarehouseResourceGateway.PalletBatchCompensation(
-                        preparation.warehouseId(), command.getBatchKey(), "COMPENSATE:" + command.getId(),
-                        "Lastleg local transaction did not commit"));
-                } catch (RuntimeException compensationFailure) {
-                    failure.addSuppressed(compensationFailure);
+                ContainerPlanConversionCommand current = commands.findByBooking(bookingId);
+                if (current != null && "COMPLETED".equals(current.getStatus())) {
+                    return result(current);
+                }
+                if (current != null && "LOCAL_COMMITTED".equals(current.getStatus())) {
+                    command.setStatus(current.getStatus());
+                    command.setResultJson(current.getResultJson());
+                    localResult = result(current);
+                } else {
+                    compensateAndFail(command, preparation, failure);
+                    throw failure;
                }
-                resetForRetry(command, failure);
-                throw failure;
            }
        }

@@ -136,6 +162,35 @@ public class ContainerInboundPlanConversionCoordinator {
    private ContainerPlanConversionCommand createCommand(String bookingId, String key,
                                                          ConvertContainerInboundPlanRequest request,
                                                          String requestHash) {
+        long commandId = IdWorker.getId();
+        ContainerPlanConversionCommand command = prepareCommand(commandId, commandId, bookingId, key, request, requestHash);
+        try { inTransaction(() -> { commands.insert(command); return null; }); }
+        catch (DuplicateKeyException concurrent) {
+            ContainerPlanConversionCommand existing = commands.findByBooking(bookingId);
+            if (existing == null) throw concurrent;
+            return existing;
+        }
+        return command;
+    }
+
+    private ContainerPlanConversionCommand replaceFailedCommand(ContainerPlanConversionCommand previous,
+                                                                  String bookingId, String key,
+                                                                  ConvertContainerInboundPlanRequest request,
+                                                                  String requestHash) {
+        ContainerPlanConversionCommand replacement = prepareCommand(previous.getId(), IdWorker.getId(),
+            bookingId, key, request, requestHash);
+        int updated = commands.replaceFailed(previous.getId(), key, requestHash, request.snapshotVersion(),
+            request.snapshotHash(), replacement.getWarehouseId(), replacement.getBatchKey(),
+            replacement.getPreparationJson(), replacement.getUpdatedAt());
+        if (updated != 1) {
+            throw new ResponseStatusException(HttpStatus.CONFLICT, "该柜正在刷新入库计划，请勿重复提交");
+        }
+        return replacement;
+    }
+
+    private ContainerPlanConversionCommand prepareCommand(long commandId, long generationId, String bookingId,
+                                                            String key, ConvertContainerInboundPlanRequest request,
+                                                            String requestHash) {
        ContainerInboundPlanPreviewResponse preview = plans.previewContainer(bookingId);
        if (preview.snapshotVersion() != request.snapshotVersion()
            || !Objects.equals(preview.snapshotHash(), request.snapshotHash()) || !preview.ready()) {
@@ -146,7 +201,6 @@ public class ContainerInboundPlanConversionCoordinator {
        catch (RuntimeException exception) {
            throw new ResponseStatusException(HttpStatus.CONFLICT, "柜级接柜仓 ID 无效", exception);
        }
-        long commandId = IdWorker.getId();
        List<PreparedItem> items = new ArrayList<>();
        int globalSequence = 0;
        for (ContainerInboundPlanPreviewResponse.Order order : preview.groups().stream()
@@ -155,7 +209,7 @@ public class ContainerInboundPlanConversionCoordinator {
            String planNo = numbers.next("INBOUND_PLAN", order.destinationWarehouseCode(), planId);
            for (int localSequence = 1; localSequence <= order.suggestedPalletCount(); localSequence++) {
                items.add(new PreparedItem(order.customerId(), Long.parseLong(order.orderId()), planId, planNo,
-                    localSequence, ++globalSequence, "CIP:" + commandId + ":" + globalSequence,
+                    localSequence, ++globalSequence, "CIP:" + generationId + ":" + globalSequence,
                    "FBA".equals(order.destinationType()) ? order.destinationWarehouseCode() : null));
            }
        }
@@ -164,15 +218,9 @@ public class ContainerInboundPlanConversionCoordinator {
        command.setId(commandId); command.setSourceContainerBookingId(bookingId); command.setIdempotencyKey(key);
        command.setRequestHash(requestHash); command.setSnapshotVersion(request.snapshotVersion());
        command.setSnapshotHash(request.snapshotHash()); command.setWarehouseId(warehouseId);
-        command.setBatchKey("CIP:" + hash(bookingId) + ":" + commandId); command.setStatus("PREPARED");
+        command.setBatchKey("CIP:" + hash(bookingId) + ":" + generationId); command.setStatus("PREPARED");
        command.setPreparationJson(write(preparation)); command.setCreatedAt(LocalDateTime.now());
        command.setUpdatedAt(command.getCreatedAt());
-        try { inTransaction(() -> { commands.insert(command); return null; }); }
-        catch (DuplicateKeyException concurrent) {
-            ContainerPlanConversionCommand existing = commands.findByBooking(bookingId);
-            if (existing == null) throw concurrent;
-            return existing;
-        }
        return command;
    }

@@ -213,11 +261,26 @@ public class ContainerInboundPlanConversionCoordinator {

    private void requireBatch(WarehouseResourceGateway.PalletBatchResponse response, String batchKey, int size,
                              String... acceptedStatuses) {
-        if (response == null || !Objects.equals(batchKey, response.batchKey()) || response.items() == null
+        if (response == null || response.batchId() <= 0 || response.attemptNo() <= 0
+            || !Objects.equals(batchKey, response.batchKey()) || response.items() == null
            || response.items().size() != size || java.util.Arrays.stream(acceptedStatuses)
                .noneMatch(status -> status.equals(response.status()))) {
            throw new ResponseStatusException(HttpStatus.CONFLICT, "海外仓托盘批次响应不完整或状态不一致");
        }
+        String expectedItemStatus = switch (response.status()) {
+            case "RESERVED" -> "RESERVED";
+            case "CONFIRMED" -> "CONFIRMED";
+            case "COMPENSATED" -> "RELEASED";
+            default -> null;
+        };
+        boolean invalidItem = expectedItemStatus == null || response.items().stream().anyMatch(item ->
+            item.customerId() <= 0 || item.sequenceNo() <= 0 || item.palletId() <= 0 || item.locationId() <= 0
+                || item.resourceLockId() <= 0 || trimToNull(item.businessKey()) == null
+                || trimToNull(item.inboundPlanNo()) == null || trimToNull(item.palletCode()) == null
+                || item.palletCode().length() > 128 || !expectedItemStatus.equals(item.status()));
+        if (invalidItem) {
+            throw new ResponseStatusException(HttpStatus.CONFLICT, "海外仓托盘批次明细不完整或状态不一致");
+        }
    }

    private void requireSameRequest(ContainerPlanConversionCommand command, String requestHash) {
@@ -235,19 +298,41 @@ public class ContainerInboundPlanConversionCoordinator {
    }

    private void complete(ContainerPlanConversionCommand command, List<InboundPlanResponse> result) {
-        inTransaction(() -> { state(command.getId(), "LOCAL_COMMITTED", "COMPLETED", write(result), null); return null; });
+        inTransaction(() -> { state(command, "LOCAL_COMMITTED", "COMPLETED", write(result), null); return null; });
        command.setStatus("COMPLETED");
    }

-    private void resetForRetry(ContainerPlanConversionCommand command, RuntimeException failure) {
+    private void failWithoutResources(ContainerPlanConversionCommand command, RuntimeException failure) {
        if ("APPLYING".equals(command.getStatus())) {
-            state(command.getId(), "APPLYING", "PREPARED", null, safeMessage(failure));
-            command.setStatus("PREPARED");
+            state(command, "APPLYING", "FAILED", null, safeMessage(failure));
+            command.setStatus("FAILED");
        }
    }

-    private void state(long id, String expectedStatus, String status, String resultJson, String lastError) {
-        if (commands.updateState(id, expectedStatus, status, resultJson, lastError, LocalDateTime.now()) != 1) {
+    private void compensateAndFail(ContainerPlanConversionCommand command, Preparation preparation,
+                                   RuntimeException failure) {
+        try {
+            WarehouseResourceGateway.PalletBatchResponse compensated = warehouse.compensatePalletBatch(
+                new WarehouseResourceGateway.PalletBatchCompensation(preparation.warehouseId(),
+                    command.getBatchKey(), "COMPENSATE:" + command.getBatchKey(),
+                    "Lastleg local transaction did not commit"));
+            requireBatch(compensated, command.getBatchKey(), preparation.items().size(), "COMPENSATED");
+            failWithoutResources(command, failure);
+        } catch (ResponseStatusException compensationFailure) {
+            if (compensationFailure.getStatusCode().value() == 404) {
+                failWithoutResources(command, failure);
+                return;
+            }
+            failure.addSuppressed(compensationFailure);
+        } catch (RuntimeException compensationFailure) {
+            failure.addSuppressed(compensationFailure);
+        }
+    }
+
+    private void state(ContainerPlanConversionCommand command, String expectedStatus, String status,
+                       String resultJson, String lastError) {
+        if (commands.updateState(command.getId(), expectedStatus, command.getLeaseToken(), status,
+            resultJson, lastError, LocalDateTime.now()) != 1) {
            throw new IllegalStateException("无法更新柜级转换命令状态");
        }
    }
@@ -281,6 +366,9 @@ public class ContainerInboundPlanConversionCoordinator {
        String message = value.getMessage();
        return message == null ? value.getClass().getSimpleName() : message.substring(0, Math.min(1000, message.length()));
    }
+    private static String trimToNull(String value) {
+        return value == null || value.isBlank() ? null : value.trim();
+    }

    public record Preparation(long warehouseId, List<PreparedItem> items) { }
    public record PreparedItem(String customerId, long orderId, long planId, String planNo,
@@ -198,7 +198,8 @@ public class InboundPlanService {
            .map(ContainerInboundPlanPreviewResponse.Order::plannedPalletCount)
            .filter(Objects::nonNull).mapToInt(Integer::intValue).sum();
        long snapshotVersion = rows.stream().mapToLong(InboundPlanOrderMapper.ContainerPreviewOrderRow::version).max().orElse(0);
-        String snapshotHash = sha256(List.of(bookingId, containerNo, warehouseId, snapshotVersion, orders));
+        String snapshotHash = sha256(java.util.Arrays.asList(
+            bookingId, containerNo, warehouseId, snapshotVersion, orders));
        return new ContainerInboundPlanPreviewResponse(bookingId, containerNo, warehouseId, warehouseCode,
            warehouseName, snapshotVersion, snapshotHash, orders.size(), orders.size() - blocked, blocked,
            pendingOrders, pendingPallets, existingPallets,