diff --git a/deploy/sql/海外区域统一增量-MySQL8.sql b/deploy/sql/海外区域统一增量-MySQL8.sql index 96ff3dd7a..e9591329d 100644 --- a/deploy/sql/海外区域统一增量-MySQL8.sql +++ b/deploy/sql/海外区域统一增量-MySQL8.sql @@ -1995,12 +1995,14 @@ CREATE TABLE IF NOT EXISTS ll_container_plan_conversion_command ( idempotency_key VARCHAR(128) NOT NULL COMMENT '首次调用幂等键', request_hash CHAR(64) NOT NULL COMMENT '确认请求SHA-256摘要', snapshot_version BIGINT NOT NULL COMMENT '确认的柜级快照版本', snapshot_hash CHAR(64) NOT NULL COMMENT '确认的柜级快照SHA-256', warehouse_id BIGINT NOT NULL COMMENT '海外接柜仓ID', batch_key VARCHAR(160) NOT NULL COMMENT '海外仓托盘批次幂等键', - status VARCHAR(32) NOT NULL COMMENT 'PREPARED/APPLYING/LOCAL_COMMITTED/COMPLETED', preparation_json JSON NOT NULL COMMENT '稳定计划号与托盘业务键快照', + status VARCHAR(32) NOT NULL COMMENT '命令状态：PREPARED/APPLYING/FAILED/LOCAL_COMMITTED/COMPLETED', lease_token VARCHAR(64) NULL COMMENT '当前执行租约令牌，用于陈旧执行者隔离', preparation_json JSON NOT NULL COMMENT '稳定计划号与托盘业务键快照', result_json JSON NULL COMMENT '本地计划提交结果', last_error VARCHAR(1000) NULL COMMENT '最近一次跨服务错误', - created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3), + created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '命令创建时间', updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3) COMMENT '命令状态或租约更新时间', PRIMARY KEY(id), UNIQUE KEY uk_ll_container_conversion_command(source_container_booking_id), UNIQUE KEY uk_ll_container_conversion_batch(warehouse_id,batch_key), KEY idx_ll_container_conversion_status(status,updated_at) ) ENGINE=InnoDB COMMENT='柜级入库计划转换Saga与幂等命令'; +SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='ll_container_plan_conversion_command' AND column_name='lease_token'), 'SELECT 1', 'ALTER TABLE ll_container_plan_conversion_command ADD COLUMN lease_token VARCHAR(64) NULL COMMENT ''当前执行租约令牌，用于陈旧执行者隔离'' AFTER status')); +PREPARE stmt FROM @ddl; EXECUTE stmt; DEALLOCATE PREPARE stmt; ALTER TABLE ll_inbound_plan_pallet MODIFY COLUMN pallet_no VARCHAR(128) NOT NULL COMMENT 'Warehouse权威实物托盘码'; SELECT 'PASS' AS migration_status, '海外尾程人工计划、暂扣与人工节点结构已就绪' AS message; diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/entity/ContainerPlanConversionCommand.java b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/entity/ContainerPlanConversionCommand.java index 327338d62..8130c3593 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/entity/ContainerPlanConversionCommand.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/entity/ContainerPlanConversionCommand.java @@ -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; diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/mapper/ContainerPlanConversionCommandMapper.java b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/mapper/ContainerPlanConversionCommandMapper.java index 8ddca8d0b..a4bc87cc3 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/mapper/ContainerPlanConversionCommandMapper.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/mapper/ContainerPlanConversionCommandMapper.java @@ -17,17 +17,33 @@ public interface ContainerPlanConversionCommandMapper extends BaseMapper existing = inTransaction(() -> { List 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 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 preparedPlans = bind(preparation, reservation); localResult = inTransaction(() -> { List 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 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 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 items) { } public record PreparedItem(String customerId, long orderId, long planId, String planNo, diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/service/InboundPlanService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/service/InboundPlanService.java index b2ab343e1..bae48b5c6 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/service/InboundPlanService.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/service/InboundPlanService.java @@ -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, diff --git a/lastleg-service/src/main/resources/db/lastleg.sql b/lastleg-service/src/main/resources/db/lastleg.sql index 49374729d..2712d3cbc 100644 --- a/lastleg-service/src/main/resources/db/lastleg.sql +++ b/lastleg-service/src/main/resources/db/lastleg.sql @@ -587,17 +587,20 @@ CREATE TABLE IF NOT EXISTS ll_container_plan_conversion_command ( snapshot_hash CHAR(64) NOT NULL COMMENT '确认的柜级快照SHA-256', warehouse_id BIGINT NOT NULL COMMENT '海外接柜仓ID', batch_key VARCHAR(160) NOT NULL COMMENT '海外仓托盘批次幂等键', - status VARCHAR(32) NOT NULL COMMENT 'PREPARED/APPLYING/LOCAL_COMMITTED/COMPLETED', + status VARCHAR(32) NOT NULL COMMENT '命令状态：PREPARED/APPLYING/FAILED/LOCAL_COMMITTED/COMPLETED', + lease_token VARCHAR(64) NULL COMMENT '当前执行租约令牌，用于陈旧执行者隔离', preparation_json JSON NOT NULL COMMENT '稳定计划号与托盘业务键快照', result_json JSON NULL COMMENT '本地计划提交结果', last_error VARCHAR(1000) NULL COMMENT '最近一次跨服务错误', - created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), - updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3), + created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '命令创建时间', + updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3) COMMENT '命令状态或租约更新时间', PRIMARY KEY(id), UNIQUE KEY uk_ll_container_conversion_command(source_container_booking_id), UNIQUE KEY uk_ll_container_conversion_batch(warehouse_id,batch_key), KEY idx_ll_container_conversion_status(status,updated_at) ) ENGINE=InnoDB COMMENT='柜级入库计划转换Saga与幂等命令'; +SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='ll_container_plan_conversion_command' AND column_name='lease_token'), 'SELECT 1', 'ALTER TABLE ll_container_plan_conversion_command ADD COLUMN lease_token VARCHAR(64) NULL COMMENT ''当前执行租约令牌，用于陈旧执行者隔离'' AFTER status')); +PREPARE container_conversion_lease_stmt FROM @ddl; EXECUTE container_conversion_lease_stmt; DEALLOCATE PREPARE container_conversion_lease_stmt; CREATE TABLE IF NOT EXISTS ll_inbound_plan_line ( id BIGINT NOT NULL COMMENT '入库计划明细ID（雪花ID）', diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/service/ContainerInboundPlanConversionCoordinatorTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/service/ContainerInboundPlanConversionCoordinatorTest.java index 196833bd4..0e8b2736d 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/service/ContainerInboundPlanConversionCoordinatorTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/service/ContainerInboundPlanConversionCoordinatorTest.java @@ -1,9 +1,11 @@ package com.baiyue.lastleg.inboundplan.service; import com.baiyue.lastleg.inboundplan.dto.ConvertContainerInboundPlanRequest; +import com.baiyue.lastleg.inboundplan.dto.ContainerInboundPlanPreviewResponse; import com.baiyue.lastleg.inboundplan.entity.ContainerPlanConversionCommand; import com.baiyue.lastleg.inboundplan.mapper.ContainerPlanConversionCommandMapper; import com.baiyue.lastleg.inboundplan.pda.gateway.WarehouseResourceGateway; +import com.baiyue.lastleg.service.LastlegOrderNoGenerator; import com.fasterxml.jackson.databind.json.JsonMapper; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -11,8 +13,12 @@ import org.junit.jupiter.api.extension.ExtendWith; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.support.SimpleTransactionStatus; import org.springframework.web.server.ResponseStatusException; +import java.time.LocalDateTime; +import java.math.BigDecimal; import java.util.List; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -20,6 +26,8 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import static org.mockito.Mockito.any; +import static org.mockito.Mockito.eq; @ExtendWith(MockitoExtension.class) class ContainerInboundPlanConversionCoordinatorTest { @@ -29,6 +37,8 @@ class ContainerInboundPlanConversionCoordinatorTest { @Mock InboundPlanService plans; @Mock WarehouseResourceGateway warehouse; @Mock ContainerPlanConversionCommandMapper commands; + @Mock LastlegOrderNoGenerator numbers; + @Mock PlatformTransactionManager transactionManager; private ContainerInboundPlanConversionCoordinator coordinator; private JsonMapper objectMapper; @@ -40,7 +50,11 @@ class ContainerInboundPlanConversionCoordinatorTest { ReflectionTestUtils.setField(coordinator, "plans", plans); ReflectionTestUtils.setField(coordinator, "warehouse", warehouse); ReflectionTestUtils.setField(coordinator, "commands", commands); + ReflectionTestUtils.setField(coordinator, "numbers", numbers); ReflectionTestUtils.setField(coordinator, "objectMapper", objectMapper); + ReflectionTestUtils.setField(coordinator, "transactionManager", transactionManager); + org.mockito.Mockito.lenient().when(transactionManager.getTransaction(any())) + .thenReturn(new SimpleTransactionStatus()); } @Test @@ -57,6 +71,19 @@ class ContainerInboundPlanConversionCoordinatorTest { verify(commands, never()).insert(org.mockito.ArgumentMatchers.any(ContainerPlanConversionCommand.class)); } + @Test + void sameIdempotencyKeyStillRejectsAlteredCompletedRequest() throws Exception { + ContainerPlanConversionCommand completed = command("FIRST-KEY", "COMPLETED", request()); + completed.setResultJson("[]"); + when(commands.find(BOOKING_ID, "FIRST-KEY")).thenReturn(completed); + + ConvertContainerInboundPlanRequest changed = + new ConvertContainerInboundPlanRequest(BOOKING_ID, 8L, "b".repeat(64)); + + assertEquals(409, assertThrows(ResponseStatusException.class, + () -> coordinator.confirm(changed, "FIRST-KEY")).getStatusCode().value()); + } + @Test void concurrentCabinetClaimRejectsFollowerBeforeWarehouseReservation() throws Exception { ConvertContainerInboundPlanRequest request = request(); @@ -78,7 +105,7 @@ class ContainerInboundPlanConversionCoordinatorTest { when(commands.find(BOOKING_ID, "SECOND-KEY")).thenReturn(null); when(commands.findByBooking(BOOKING_ID)).thenReturn(prepared); when(commands.claim(org.mockito.ArgumentMatchers.eq(1L), org.mockito.ArgumentMatchers.eq("PREPARED"), - org.mockito.ArgumentMatchers.eq("APPLYING"), org.mockito.ArgumentMatchers.any())).thenReturn(0); + org.mockito.ArgumentMatchers.eq("APPLYING"), any(String.class), org.mockito.ArgumentMatchers.any())).thenReturn(0); ResponseStatusException error = assertThrows(ResponseStatusException.class, () -> coordinator.confirm(request, "SECOND-KEY")); @@ -102,6 +129,102 @@ class ContainerInboundPlanConversionCoordinatorTest { assertEquals(409, error.getStatusCode().value()); } + @Test + void staleApplyingLeaseIsReclaimedWithoutRepeatingPreparedClaim() throws Exception { + ConvertContainerInboundPlanRequest request = request(); + ContainerPlanConversionCommand stale = command("FIRST-KEY", "APPLYING", request); + stale.setUpdatedAt(LocalDateTime.now().minusMinutes(3)); + when(commands.find(BOOKING_ID, "FIRST-KEY")).thenReturn(stale); + when(commands.reclaimStale(eq(1L), eq(stale.getUpdatedAt()), any(String.class), any())).thenReturn(1); + when(plans.convertContainerReserved(BOOKING_ID, 7L, SNAPSHOT_HASH, List.of())).thenReturn(List.of()); + when(commands.updateState(eq(1L), eq("APPLYING"), any(String.class), eq("COMPLETED"), eq("[]"), + org.mockito.ArgumentMatchers.isNull(), any())).thenReturn(1); + + assertEquals(List.of(), coordinator.confirm(request, "FIRST-KEY")); + + verify(commands, never()).claim(any(Long.class), any(String.class), any(String.class), + any(String.class), any()); + } + + @Test + void compensatedFailureCanBeReplacedByFreshSnapshotAndComplete() throws Exception { + ContainerPlanConversionCommand failed = command("OLD-KEY", "FAILED", request()); + when(commands.find(BOOKING_ID, "NEW-KEY")).thenReturn(null); + when(commands.findByBooking(BOOKING_ID)).thenReturn(failed); + when(plans.previewContainer(BOOKING_ID)).thenReturn(preview(8L, "b".repeat(64))); + when(commands.replaceFailed(eq(1L), eq("NEW-KEY"), any(), eq(8L), eq("b".repeat(64)), + eq(31L), any(), any(), any())).thenReturn(1); + when(commands.claim(eq(1L), eq("PREPARED"), eq("APPLYING"), any(String.class), any())).thenReturn(1); + when(plans.convertContainerReserved(BOOKING_ID, 8L, "b".repeat(64), List.of())).thenReturn(List.of()); + when(commands.updateState(eq(1L), eq("APPLYING"), any(String.class), eq("COMPLETED"), eq("[]"), + org.mockito.ArgumentMatchers.isNull(), any())).thenReturn(1); + + assertEquals(List.of(), coordinator.confirm( + new ConvertContainerInboundPlanRequest(BOOKING_ID, 8L, "b".repeat(64)), "NEW-KEY")); + } + + @Test + void invalidEnsureResponseIsCompensatedBeforeCommandBecomesFailed() throws Exception { + ConvertContainerInboundPlanRequest request = request(); + ContainerPlanConversionCommand prepared = command("FIRST-KEY", "PREPARED", request); + setOnePreparedItem(prepared); + when(commands.find(BOOKING_ID, "FIRST-KEY")).thenReturn(prepared); + when(commands.claim(eq(1L), eq("PREPARED"), eq("APPLYING"), any(String.class), any())).thenReturn(1); + when(warehouse.ensurePalletBatch(any())).thenReturn(new WarehouseResourceGateway.PalletBatchResponse( + 0L, "BATCH-1", "RESERVED", 1, false, List.of(reserved("RESERVED")))); + when(warehouse.compensatePalletBatch(any())).thenReturn(new WarehouseResourceGateway.PalletBatchResponse( + 41L, "BATCH-1", "COMPENSATED", 1, false, List.of(reserved("RELEASED")))); + when(commands.updateState(eq(1L), eq("APPLYING"), any(String.class), eq("FAILED"), + org.mockito.ArgumentMatchers.isNull(), any(), any())).thenReturn(1); + + assertEquals(409, assertThrows(ResponseStatusException.class, + () -> coordinator.confirm(request, "FIRST-KEY")).getStatusCode().value()); + + verify(warehouse).compensatePalletBatch(any()); + } + + @Test + void uncertainLocalCommitIsRereadBeforeDecidingWhetherToCompensate() throws Exception { + ConvertContainerInboundPlanRequest request = request(); + ContainerPlanConversionCommand prepared = command("FIRST-KEY", "PREPARED", request); + setOnePreparedItem(prepared); + ContainerPlanConversionCommand committed = command("FIRST-KEY", "LOCAL_COMMITTED", request); + committed.setResultJson("[]"); + when(commands.find(BOOKING_ID, "FIRST-KEY")).thenReturn(prepared); + when(commands.claim(eq(1L), eq("PREPARED"), eq("APPLYING"), any(String.class), any())).thenReturn(1); + when(warehouse.ensurePalletBatch(any())).thenReturn(new WarehouseResourceGateway.PalletBatchResponse( + 41L, "BATCH-1", "RESERVED", 1, false, List.of(reserved("RESERVED")))); + when(plans.convertContainerReserved(eq(BOOKING_ID), eq(7L), eq(SNAPSHOT_HASH), any())) + .thenThrow(new IllegalStateException("commit outcome unknown")); + when(commands.findByBooking(BOOKING_ID)).thenReturn(committed); + when(warehouse.confirmPalletBatch(any())).thenReturn(new WarehouseResourceGateway.PalletBatchResponse( + 41L, "BATCH-1", "CONFIRMED", 1, false, List.of(reserved("CONFIRMED")))); + when(commands.updateState(eq(1L), eq("LOCAL_COMMITTED"), any(String.class), eq("COMPLETED"), eq("[]"), + org.mockito.ArgumentMatchers.isNull(), any())).thenReturn(1); + + assertEquals(List.of(), coordinator.confirm(request, "FIRST-KEY")); + + verify(warehouse, never()).compensatePalletBatch(any()); + } + + @Test + void localCommittedReplayOnlyRetriesRemoteConfirmation() throws Exception { + ConvertContainerInboundPlanRequest request = request(); + ContainerPlanConversionCommand committed = command("FIRST-KEY", "LOCAL_COMMITTED", request); + committed.setResultJson("[]"); + setOnePreparedItem(committed); + when(commands.find(BOOKING_ID, "FIRST-KEY")).thenReturn(committed); + when(warehouse.confirmPalletBatch(any())).thenReturn(new WarehouseResourceGateway.PalletBatchResponse( + 41L, "BATCH-1", "CONFIRMED", 1, false, List.of(reserved("CONFIRMED")))); + when(commands.updateState(eq(1L), eq("LOCAL_COMMITTED"), any(String.class), eq("COMPLETED"), eq("[]"), + org.mockito.ArgumentMatchers.isNull(), any())).thenReturn(1); + + assertEquals(List.of(), coordinator.confirm(request, "FIRST-KEY")); + + verify(warehouse, never()).ensurePalletBatch(any()); + verify(warehouse).confirmPalletBatch(any()); + } + private ConvertContainerInboundPlanRequest request() { return new ConvertContainerInboundPlanRequest(BOOKING_ID, 7L, SNAPSHOT_HASH); } @@ -119,8 +242,26 @@ class ContainerInboundPlanConversionCoordinatorTest { command.setWarehouseId(31L); command.setBatchKey("BATCH-1"); command.setStatus(status); + if ("APPLYING".equals(status) || "LOCAL_COMMITTED".equals(status)) command.setLeaseToken("LEASE-1"); command.setPreparationJson(objectMapper.writeValueAsString( new ContainerInboundPlanConversionCoordinator.Preparation(31L, List.of()))); return command; } + + private void setOnePreparedItem(ContainerPlanConversionCommand command) throws Exception { + var item = new ContainerInboundPlanConversionCoordinator.PreparedItem( + "100", 11L, 21L, "IP-21", 1, 1, "CIP:1:1", "ONT8"); + command.setPreparationJson(objectMapper.writeValueAsString( + new ContainerInboundPlanConversionCoordinator.Preparation(31L, List.of(item)))); + } + + private WarehouseResourceGateway.PalletBatchItem reserved(String status) { + return new WarehouseResourceGateway.PalletBatchItem( + 100L, "CIP:1:1", "IP-21", 1, 51L, "PLT-51", 61L, 71L, status); + } + + private ContainerInboundPlanPreviewResponse preview(long version, String hash) { + return new ContainerInboundPlanPreviewResponse(BOOKING_ID, "MSCU6639870", "31", "US-LAX-01", + "Los Angeles", version, hash, 1, 1, 0, 0, 0, 1, 1, 1, true, List.of()); + } } diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/service/ContainerInboundPlanPreviewTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/service/ContainerInboundPlanPreviewTest.java index fb7c97688..b5186dec5 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/service/ContainerInboundPlanPreviewTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/service/ContainerInboundPlanPreviewTest.java @@ -67,6 +67,21 @@ class ContainerInboundPlanPreviewTest { assertEquals(List.of("DESTINATION_NOT_READY"), result.groups().get(4).orders().get(0).blockers()); } + @Test + void hashesBlockedPreviewEvenWhenContainerAndWarehouseFactsAreNull() { + when(orderMapper.containerPreviewOrders("BOOKING-NULL")).thenReturn(List.of( + new InboundPlanOrderMapper.ContainerPreviewOrderRow("100", 1L, "LM-1", "FO-1", "测试客户", + "FIRST_LEG", "OVERSEAS", "IN-1", "PENDING_ASSIGNMENT", null, null, 0, 0, + BigDecimal.ZERO, null, null, null, null, null, null, null, 0, false, 1, + null, null, null))); + + var result = service.previewContainer("BOOKING-NULL"); + + assertNotNull(result.snapshotHash()); + assertEquals(64, result.snapshotHash().length()); + assertFalse(result.ready()); + } + private InboundPlanOrderMapper.ContainerPreviewOrderRow row( long id, String orderNo, String orderType, String destinationType, String destinationCode, String destinationName, int recommendedPallets, String planNo, Integer plannedPallets, int version) {