diff --git a/firstleg/src/main/java/com/baiyue/firstleg/loading/service/LoadingDeliveryService.java b/firstleg/src/main/java/com/baiyue/firstleg/loading/service/LoadingDeliveryService.java index 8e97acbdc..aa1795907 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/loading/service/LoadingDeliveryService.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/loading/service/LoadingDeliveryService.java @@ -116,13 +116,23 @@ public class LoadingDeliveryService { private LoadingActionResponse completeLoading(long tenantId, long inboundId, long evidenceInboundId, long operatorId, String operatorName, String key, CompleteLoadingRequest request) { + return completeLoading(tenantId, inboundId, evidenceInboundId, operatorId, operatorName, + key, request, false); + } + + private LoadingActionResponse completeLoading(long tenantId, long inboundId, long evidenceInboundId, + long operatorId, String operatorName, String key, + CompleteLoadingRequest request, + boolean acceptedPdaCompletion) { String hash = hash(request); - LoadingActionResponse replay = claimOrReplay(tenantId, inboundId, operatorId, COMPLETE_LOADING, key, hash); + LoadingActionResponse replay = claimOrReplay(tenantId, inboundId, operatorId, COMPLETE_LOADING, + key, hash, acceptedPdaCompletion); if (replay != null) return replay; Snapshot inbound = requireInbound(tenantId, inboundId); CompletionTarget target = repository.completionTarget(tenantId, inboundId); if (target != null) { - return completeFulfillmentSlice(tenantId, evidenceInboundId, operatorId, operatorName, key, request, target); + return completeFulfillmentSlice(tenantId, evidenceInboundId, operatorId, operatorName, + key, request, target, acceptedPdaCompletion); } if ("LOADED".equals(inbound.processStatus()) || "DELIVERED".equals(inbound.processStatus())) { loadingModelBridge.requireCompleted(tenantId, inboundId); @@ -132,7 +142,7 @@ public class LoadingDeliveryService { if (!"LOADING".equals(inbound.processStatus())) throw new BizException(ResultCode.CONFLICT, "当前订单不在装柜中"); loadingModelBridge.requireCompleted(tenantId, inboundId); completionGate.requireReleasedBooking(tenantId, inboundId); - requireCompletedPicking(tenantId, inboundId); + requirePickingGate(tenantId, inboundId, acceptedPdaCompletion); requireCompletionEvidence(tenantId, evidenceInboundId); if (repository.completeLoading(tenantId, inboundId, operatorId, request.expectedVersion()) != 1) throw concurrent(); addHistory(tenantId, inboundId, "LOADING", "LOADED", COMPLETE_LOADING, request.remark(), operatorId, operatorName); @@ -145,12 +155,14 @@ public class LoadingDeliveryService { * Resolve and lock the canonical completion row after settlement, then retain optimistic locking * with that current server version instead of reusing the pre-settlement task snapshot. */ + @Transactional public LoadingActionResponse completeAfterPdaSettlement(long tenantId, long inboundId, long operatorId, String operatorName, String key, String remark) { return completeAfterPdaSettlement(tenantId, inboundId, inboundId, operatorId, operatorName, key, remark); } + @Transactional public LoadingActionResponse completeAfterPdaSettlement(long tenantId, long inboundId, long evidenceInboundId, long operatorId, String operatorName, String key, String remark) { @@ -161,7 +173,7 @@ public class LoadingDeliveryService { throw new BizException(ResultCode.CONFLICT, "PDA结算后的实装履约分片不存在，请刷新后重试"); } return completeLoading(tenantId, inboundId, evidenceInboundId, operatorId, operatorName, key, - new CompleteLoadingRequest(current.version(), remark)); + new CompleteLoadingRequest(current.version(), remark), true); } /** @@ -185,7 +197,8 @@ public class LoadingDeliveryService { private LoadingActionResponse completeFulfillmentSlice(long tenantId, long evidenceInboundId, long operatorId, String operatorName, String key, CompleteLoadingRequest request, - CompletionTarget target) { + CompletionTarget target, + boolean acceptedPdaCompletion) { if ("LOADED".equals(target.loadedChildStatus())) { return complete(tenantId, evidenceInboundId, operatorId, COMPLETE_LOADING, key, new LoadingActionResponse("LOADED", evidenceResponse(tenantId, evidenceInboundId))); @@ -195,7 +208,7 @@ public class LoadingDeliveryService { } loadingModelBridge.requireCompleted(tenantId, target.sourceInboundId()); completionGate.requireReleasedBooking(tenantId, target.sourceInboundId()); - requireCompletedPicking(tenantId, target.sourceInboundId()); + requirePickingGate(tenantId, target.sourceInboundId(), acceptedPdaCompletion); requireCompletionEvidence(tenantId, evidenceInboundId); if (repository.completeLoadedChild(tenantId, target.sourceInboundId(), operatorId, request.expectedVersion()) != 1) throw concurrent(); @@ -308,9 +321,12 @@ public class LoadingDeliveryService { task.getExpectedCount(), task.getPickedCount(), task.getRemark()); } - private void requireCompletedPicking(long tenantId, long inboundId) { - if (!canonicalPicking(tenantId, inboundId).completed()) { - throw new BizException(ResultCode.CONFLICT, "PDA拣货复核尚未完成"); + /** completion worker 已持有 durable lease；此时 task 尚为 COMPLETING，不能循环要求最终 COMPLETED。 */ + private void requirePickingGate(long tenantId, long inboundId, boolean acceptedPdaCompletion) { + CanonicalPicking picking = canonicalPicking(tenantId, inboundId); + if (acceptedPdaCompletion ? !picking.resolved() : !picking.completed()) { + throw new BizException(ResultCode.CONFLICT, acceptedPdaCompletion + ? "PDA实装明细尚未全部解决" : "PDA拣货复核尚未完成"); } } @@ -337,9 +353,12 @@ public class LoadingDeliveryService { private record CanonicalPicking(int expectedCount, int pickedCount, int taskCount, int completedTaskCount, int unresolvedCount) { - boolean completed() { + boolean resolved() { return expectedCount > 0 && expectedCount == pickedCount && taskCount > 0 - && completedTaskCount == taskCount && unresolvedCount == 0; + && unresolvedCount == 0; + } + boolean completed() { + return resolved() && completedTaskCount == taskCount; } } @@ -431,10 +450,17 @@ public class LoadingDeliveryService { private LoadingActionResponse claimOrReplay(long tenantId, long inboundId, long operatorId, String operation, String key, String hash) { + return claimOrReplay(tenantId, inboundId, operatorId, operation, key, hash, false); + } + + private LoadingActionResponse claimOrReplay(long tenantId, long inboundId, long operatorId, + String operation, String key, String hash, + boolean resumeProcessing) { validateKey(key); if (repository.claimOperation(tenantId, inboundId, operatorId, operation, key, hash)) return null; LoadingOperationEntity existing = repository.findOperation(tenantId, inboundId, operatorId, operation, key); if (existing == null || !hash.equals(existing.getRequestHash())) throw new BizException(ResultCode.CONFLICT, "幂等键已用于其他请求"); + if (resumeProcessing && "PROCESSING".equals(existing.getStatus())) return null; if (!"SUCCESS".equals(existing.getStatus()) || !StringUtils.hasText(existing.getResultJson())) throw new BizException(ResultCode.CONFLICT, "相同请求正在处理中，请稍后重试"); try { return objectMapper.readValue(existing.getResultJson(), LoadingActionResponse.class); } catch (JsonProcessingException e) { throw new IllegalStateException("无法读取装柜幂等结果", e); } diff --git a/firstleg/src/test/java/com/baiyue/firstleg/loading/service/LoadingDeliveryServiceTest.java b/firstleg/src/test/java/com/baiyue/firstleg/loading/service/LoadingDeliveryServiceTest.java index c2e32e59b..1d140af65 100644 --- a/firstleg/src/test/java/com/baiyue/firstleg/loading/service/LoadingDeliveryServiceTest.java +++ b/firstleg/src/test/java/com/baiyue/firstleg/loading/service/LoadingDeliveryServiceTest.java @@ -25,6 +25,7 @@ import org.mockito.InjectMocks; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; import org.springframework.http.ResponseEntity; +import org.springframework.transaction.annotation.Transactional; import java.util.List; import java.util.concurrent.atomic.AtomicInteger; @@ -197,6 +198,43 @@ class LoadingDeliveryServiceTest { verify(repository, never()).completeLoadedChild(10, 20, 7, 6); } + @Test + void pdaFinalizeResumesSameProcessingClaimLeftByAnEarlierWorkerFailure() { + readyForPdaFinalizeFacts(); + CompletionTarget settled = new CompletionTarget(20L, null, 20L, + "LOADING", 7, "LOADING", 7); + when(repository.completionTarget(10, 20)).thenReturn(settled); + when(repository.findInboundForUpdate(10, 20)).thenReturn(new Snapshot(20L, 10L, "LOADING", 7)); + when(repository.findInbound(10, 20)).thenReturn(new Snapshot(20L, 10L, "LOADING", 7)); + when(repository.completeLoadedChild(10, 20, 7, 7)).thenReturn(1); + AtomicReference hash = new AtomicReference<>(); + when(repository.claimOperation(eq(10L), eq(20L), eq(7L), eq("COMPLETE_LOADING"), + eq("pda-resume"), anyString())).thenAnswer(invocation -> { + hash.set(invocation.getArgument(5)); + return false; + }); + when(repository.findOperation(10L, 20L, 7L, "COMPLETE_LOADING", "pda-resume")) + .thenAnswer(invocation -> processingOperation(hash.get())); + + LoadingActionResponse result = service.completeAfterPdaSettlement( + 10, 20, 7, "admin", "pda-resume", "完成"); + + assertThat(result.mainStatus()).isEqualTo("LOADED"); + verify(repository).completeLoadedChild(10, 20, 7, 7); + verify(repository).completeOperation(eq(10L), eq(20L), eq(7L), + eq("COMPLETE_LOADING"), eq("pda-resume"), anyString()); + } + + @Test + void pdaFinalizeEntryPointsAreTransactional() throws Exception { + assertThat(LoadingDeliveryService.class.getMethod("completeAfterPdaSettlement", + long.class, long.class, long.class, String.class, String.class, String.class) + .getAnnotation(Transactional.class)).isNotNull(); + assertThat(LoadingDeliveryService.class.getMethod("completeAfterPdaSettlement", + long.class, long.class, long.class, long.class, String.class, String.class, String.class) + .getAnnotation(Transactional.class)).isNotNull(); + } + @Test void pdaFinalizeLocksAndUsesLoadedSliceVersionAfterPartialSettlement() { readyForPdaFinalize(); @@ -436,11 +474,16 @@ class LoadingDeliveryServiceTest { private void readyForPdaFinalize() { when(repository.claimOperation(anyLong(), anyLong(), anyLong(), anyString(), anyString(), anyString())) .thenReturn(true); + readyForPdaFinalizeFacts(); + } + + private void readyForPdaFinalizeFacts() { LoadingPickItemMapper.InboundProgress progress = mock(LoadingPickItemMapper.InboundProgress.class); when(progress.getExpectedCount()).thenReturn(5); when(progress.getPickedCount()).thenReturn(5); when(progress.getTaskCount()).thenReturn(1); - when(progress.getCompletedTaskCount()).thenReturn(1); + // completion worker runs before task finalization; accepted COMPLETING is not COMPLETED yet. + when(progress.getCompletedTaskCount()).thenReturn(0); when(progress.getUnresolvedCount()).thenReturn(0); when(loadingPicks.inboundProgress(10, 20)).thenReturn(progress); when(repository.listEvidence(10, 20)).thenReturn(List.of( @@ -462,4 +505,11 @@ class LoadingDeliveryServiceTest { value.setResultJson(resultJson); return value; } + + private static LoadingOperationEntity processingOperation(String requestHash) { + LoadingOperationEntity value = new LoadingOperationEntity(); + value.setRequestHash(requestHash); + value.setStatus("PROCESSING"); + return value; + } }