diff --git a/firstleg/src/main/java/com/baiyue/firstleg/mapper/LoadingCompletedEventMapper.java b/firstleg/src/main/java/com/baiyue/firstleg/mapper/LoadingCompletedEventMapper.java index f849fa865..60f52f5a7 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/mapper/LoadingCompletedEventMapper.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/mapper/LoadingCompletedEventMapper.java @@ -43,9 +43,7 @@ public interface LoadingCompletedEventMapper { AND actual_task.loading_plan_id = p.id JOIN fl_inbound_order o ON o.id = i.inbound_id AND o.tenant_id = i.account_id AND o.deleted = 0 LEFT JOIN fl_inbound_order root_order ON root_order.id = COALESCE(o.root_inbound_id, o.id) - WHERE p.id = #{planId} AND p.plan_status = 'COMPLETED' - AND o.process_status = 'LOADED' - AND i.package_count > 0 + WHERE p.id = #{planId} AND i.package_count > 0 ORDER BY i.account_id, i.inbound_id, i.inbound_detail_id, i.id """) List rows(@Param("planId") long planId); diff --git a/firstleg/src/main/java/com/baiyue/firstleg/service/LoadingCompletedOutboxService.java b/firstleg/src/main/java/com/baiyue/firstleg/service/LoadingCompletedOutboxService.java index aabcf8567..c99cdddc7 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/service/LoadingCompletedOutboxService.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/service/LoadingCompletedOutboxService.java @@ -48,16 +48,26 @@ public class LoadingCompletedOutboxService { @Resource private ObjectMapper objectMapper; public void publish(long planId) { + preparedEnvelopes(planId).forEach(this::insert); + } + + /** 在任何入库结算前验证跨区交接快照,避免整托/箱唛冲突留下半完成状态。 */ + public void requirePublishable(long planId) { + preparedEnvelopes(planId); + } + + private List preparedEnvelopes(long planId) { List rows = snapshotMapper.rows(planId); if (rows.isEmpty()) throw new BizException(ResultCode.CONFLICT, "装柜计划没有最终实装入仓单,不能生成尾程交接事件"); validateContainerIdentity(rows); validateCustomerSnapshots(rows); Map> cartonMarks = cartonLabelPools(rows); List palletRows = palletRows(planId); - rows.stream().collect(java.util.stream.Collectors.groupingBy(Row::getAccountId, + return rows.stream().collect(java.util.stream.Collectors.groupingBy(Row::getAccountId, java.util.TreeMap::new, java.util.stream.Collectors.toList())) - .forEach((accountId, accountRows) -> insert(envelope(accountId, accountRows, - palletRows.stream().filter(item -> accountId.equals(item.getAccountId())).toList(), cartonMarks))); + .entrySet().stream().map(entry -> envelope(entry.getKey(), entry.getValue(), + palletRows.stream().filter(item -> entry.getKey().equals(item.getAccountId())).toList(), cartonMarks)) + .toList(); } /** 已开展逐箱打托的入仓单使用现场事实;未开展的存量入仓单继续兼容原托盘分配快照。 */ diff --git a/firstleg/src/main/java/com/baiyue/firstleg/service/LoadingListService.java b/firstleg/src/main/java/com/baiyue/firstleg/service/LoadingListService.java index 44e95aa08..762a479ba 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/service/LoadingListService.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/service/LoadingListService.java @@ -571,6 +571,12 @@ public class LoadingListService { loadingCompletedOutboxService.publish(planId); } + /** 完成 worker 在远程结算前验证最终跨区快照能够完整生成。 */ + @Transactional(readOnly = true) + public void requireCompletedEventReady(long planId) { + loadingCompletedOutboxService.requirePublishable(planId); + } + private AggregateRow requirePlan(long planId) { AggregateRow row = repository.aggregate(planId); if (row == null) { diff --git a/firstleg/src/main/java/com/baiyue/firstleg/service/LoadingPickService.java b/firstleg/src/main/java/com/baiyue/firstleg/service/LoadingPickService.java index 3ce02627b..befffc09f 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/service/LoadingPickService.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/service/LoadingPickService.java @@ -535,6 +535,8 @@ public class LoadingPickService { List taskGroups = coloadGroups(owners); runCompletionStage("CABINET_EVIDENCE", () -> cabinetEvidenceService.requireComplete(task.getContainerBookingId())); + runCompletionStage("COMPLETED_EVENT_READY", () -> + loadingListService.requireCompletedEventReady(task.getLoadingPlanId())); // Phase 1 is read-only: cabinet evidence is checked once, then every owner must pass its business gates // before any warehouse resource or inbound status can be changed. This prevents a later customer gate from diff --git a/firstleg/src/test/java/com/baiyue/firstleg/service/LoadingCompletedOutboxServiceTest.java b/firstleg/src/test/java/com/baiyue/firstleg/service/LoadingCompletedOutboxServiceTest.java index e76c3cd2c..dd2013b77 100644 --- a/firstleg/src/test/java/com/baiyue/firstleg/service/LoadingCompletedOutboxServiceTest.java +++ b/firstleg/src/test/java/com/baiyue/firstleg/service/LoadingCompletedOutboxServiceTest.java @@ -234,7 +234,7 @@ class LoadingCompletedOutboxServiceTest { assertThat(firstMarks).doesNotContainAnyElementsOf(secondMarks); } - @Test void snapshotOnlyPublishesFinalizedLoadedInbounds() throws Exception { + @Test void snapshotSupportsReadOnlyPreflightBeforeFinalStatusMutation() throws Exception { String rowsSql = String.join(" ", LoadingCompletedEventMapper.class.getMethod("rows", long.class) .getAnnotation(Select.class).value()).replaceAll("\\s+", " "); String palletsSql = String.join(" ", LoadingCompletedEventMapper.class.getMethod("palletRows", long.class) @@ -243,8 +243,8 @@ class LoadingCompletedOutboxServiceTest { .getMethod("actualPalletRows", long.class).getAnnotation(Select.class).value()) .replaceAll("\\s+", " "); - assertThat(rowsSql).contains("p.plan_status = 'COMPLETED'") - .contains("o.process_status = 'LOADED'") + assertThat(rowsSql).doesNotContain("p.plan_status = 'COMPLETED'") + .doesNotContain("o.process_status = 'LOADED'") .contains("i.customer_name") .doesNotContain("b.customer_id AS booking_customer_id") .doesNotContain("b.customer_name") diff --git a/firstleg/src/test/java/com/baiyue/firstleg/service/LoadingPickServiceTest.java b/firstleg/src/test/java/com/baiyue/firstleg/service/LoadingPickServiceTest.java index 206ad7295..ba589b1b5 100644 --- a/firstleg/src/test/java/com/baiyue/firstleg/service/LoadingPickServiceTest.java +++ b/firstleg/src/test/java/com/baiyue/firstleg/service/LoadingPickServiceTest.java @@ -966,6 +966,33 @@ class LoadingPickServiceTest { verify(loadingListService, never()).publishCompletedEvent(anyLong()); } + @Test + void unpublishableCrossRegionSnapshotFailsBeforeAnyOwnerSettlement() { + LoadingPickTaskEntity completing = task(); + completing.setTaskStatus("COMPLETING"); + when(repository.task(20L)).thenReturn(completing); + when(repository.inboundOwners(20L)).thenReturn(List.of( + new LoadingPickItemMapper.InboundOwner(101L, 50L), + new LoadingPickItemMapper.InboundOwner(101L, 60L))); + doThrow(new BizException(com.baiyue.common.core.api.ResultCode.CONFLICT, + "一件代发现场托盘箱唛数与实装箱数不一致")) + .when(loadingListService).requireCompletedEventReady(10L); + + assertThatThrownBy(() -> service.processCompletion( + completionCommand(905L, "async-unpublishable"), "worker-unpublishable")) + .isInstanceOf(LoadingPickService.CompletionProcessingException.class) + .satisfies(error -> assertThat(((LoadingPickService.CompletionProcessingException) error).stage()) + .isEqualTo("COMPLETED_EVENT_READY")); + + verify(settlementService, never()).settleOwner(anyLong(), anyLong(), anyLong(), anyString()); + verify(loadingDeliveryService, never()).completeAfterPdaSettlement(anyLong(), anyLong(), anyLong(), + anyString(), anyString(), anyString()); + verify(completionCommands, never()).checkpoint(anyLong(), anyLong(), anyLong(), anyInt(), anyString()); + verify(repository, never()).insert(any(LoadingPickOperationEntity.class)); + verify(loadingListService, never()).finishFromPda(anyLong(), anyLong(), anyLong(), anyString()); + verify(loadingListService, never()).publishCompletedEvent(anyLong()); + } + @Test void workerRestartResumesFromOwnerCheckpointsWithoutRepeatingSettlements() { assertNoTruckCoupling(); diff --git a/scripts/admin-test-env-acceptance.mjs b/scripts/admin-test-env-acceptance.mjs index 6985e2985..69439a022 100644 --- a/scripts/admin-test-env-acceptance.mjs +++ b/scripts/admin-test-env-acceptance.mjs @@ -1299,10 +1299,12 @@ export class AcceptanceRun { this.context.phase3LoadingRequireSharedOverflow = false this.context.phase3LoadingIncrementalPackageCount = 2 this.context.phase3LoadingIncrementalDetailCount = 4 - if (taxModes?.incremental) { - this.context.orderProfile = { - ...(this.context.orderProfile || {}), taxMode: taxModes.incremental, - } + this.context.orderProfile = { + ...(this.context.orderProfile || {}), + // The incremental order exercises partial/SKIP/PARTIAL mutations. Keep it STANDARD so a + // ONE_CLICK_DROPSHIP primary order remains a physically indivisible whole-pallet handover. + fulfillmentType: 'STANDARD', + ...(taxModes?.incremental ? { taxMode: taxModes.incremental } : {}), } await this.executeModule('orders-phase3-loading-incremental-fixture', () => orders.run(this)) const measurement = await import(new URL('./admin-test-env-modules/inbound-measurement.mjs', import.meta.url)) diff --git a/scripts/admin-test-env-modules/phase3-loading.mjs b/scripts/admin-test-env-modules/phase3-loading.mjs index 6f8b3367c..ccf203b13 100644 --- a/scripts/admin-test-env-modules/phase3-loading.mjs +++ b/scripts/admin-test-env-modules/phase3-loading.mjs @@ -226,15 +226,18 @@ export async function run(context) { const loadingPlan = context.apiSuccess(MODULE, '读取柜次创建时原子生成的智能装柜计划', loadingPlanResponse, '', OPTIONS) const loadingPlanId = id(loadingPlan?.id) if (!loadingPlanId) return + const initialAssignmentCount = source?.fulfillmentType === 'ONE_CLICK_DROPSHIP' + ? Number(candidate.availablePackageCount) : 1 const assignmentResponse = await request(context, 'POST', `${smartLoadingBase}/plans/${loadingPlanId}/assignments`, { - inboundId, inboundDetailId: id(candidate.inboundDetailId), packageCount: 1, + inboundId, inboundDetailId: id(candidate.inboundDetailId), packageCount: initialAssignmentCount, expectedVersion: Number(loadingPlan.version), operationId: `${context.marker}-bridge-assign`.slice(0, 128), }) const assignedLoadingPlan = context.apiSuccess(MODULE, '将本轮入库货物分配到柜次计划', assignmentResponse, '', OPTIONS) context.assert(MODULE, '柜次计划明细和版本真实持久化', assignmentResponse, - assignedLoadingPlan?.items?.some?.((item) => id(item.inboundId) === inboundId && Number(item.packageCount) === 1) + assignedLoadingPlan?.items?.some?.((item) => id(item.inboundId) === inboundId + && Number(item.packageCount) === initialAssignmentCount) && Number(assignedLoadingPlan?.version) === 1, - `loadingPlanId=${loadingPlanId}, version=${assignedLoadingPlan?.version ?? 'missing'}`, OPTIONS) + `loadingPlanId=${loadingPlanId}, assignedPackages=${initialAssignmentCount}, version=${assignedLoadingPlan?.version ?? 'missing'}`, OPTIONS) const assignedItem = assignedLoadingPlan?.items?.find?.((item) => id(item.inboundId) === inboundId) const orderRemarks = source || {} diff --git a/scripts/admin-test-env-modules/phase4-overseas-pda.mjs b/scripts/admin-test-env-modules/phase4-overseas-pda.mjs index 89db6a00e..37d93876a 100644 --- a/scripts/admin-test-env-modules/phase4-overseas-pda.mjs +++ b/scripts/admin-test-env-modules/phase4-overseas-pda.mjs @@ -279,11 +279,19 @@ async function convertReceivingContainers(context, producers) { const summaries = success(summaryResponse) && Array.isArray(summaryResponse.body?.data?.records) ? summaryResponse.body.data.records : [] const summary = summaries.find((item) => id(item?.containerNo) === containerNo) + const totalOrders = Number(summary?.totalOrderCount) + const inboundCodes = Number(summary?.inboundCodeCount) + const existingPlans = Number(summary?.planCount) + const pendingPlans = Number(summary?.pendingPlanCount) + const convertible = nonBlank(summary?.sourceContainerBookingId) + && totalOrders > 0 && inboundCodes === totalOrders + && Number(summary?.missingInboundCodeCount) === 0 + && ((existingPlans === 0 && pendingPlans === totalOrders) + || (summary?.allPlansReady === true && existingPlans === totalOrders && pendingPlans === 0)) context.addStep(MODULE, `读取待分配柜级摘要 ${containerNo}`, summaryResponse, - Boolean(summary?.eligible && nonBlank(summary?.sourceContainerBookingId) - && Number(summary?.inboundCodeCount) > 0), - `sourceContainerBookingId=${id(summary?.sourceContainerBookingId) || 'missing'}, eligible=${summary?.eligible === true}, inboundCodes=${summary?.inboundCodeCount ?? 'missing'}; ${trace(summaryResponse)}`, OPTIONS) - if (!summary?.eligible || !nonBlank(summary?.sourceContainerBookingId)) return false + convertible, + `sourceContainerBookingId=${id(summary?.sourceContainerBookingId) || 'missing'}, allPlansReady=${summary?.allPlansReady === true}, orders=${totalOrders}, inboundCodes=${inboundCodes}, plans=${existingPlans}, pending=${pendingPlans}; ${trace(summaryResponse)}`, OPTIONS) + if (!convertible) return false const body = { sourceContainerBookingId: id(summary.sourceContainerBookingId) } const key = `${context.marker}-pda-convert-${containerNo}`.slice(0, 128) const convertedResponse = await context.requestOverseas('POST', @@ -295,11 +303,11 @@ async function convertReceivingContainers(context, producers) { .flatMap((item) => Array.isArray(item?.lines) ? item.lines : []) .map((line) => id(line?.inboundCode)).filter(nonBlank)) context.addStep(MODULE, `整柜转换为逐入仓码独立计划 ${containerNo}`, convertedResponse, - plans.length === Number(summary.inboundCodeCount) + plans.length === inboundCodes && plans.every((item) => Array.isArray(item?.lines) && item.lines.length > 0) - && convertedInboundCodes.size === Number(summary.inboundCodeCount), - `plans=${plans.length}, inboundCodes=${convertedInboundCodes.size}, expected=${summary.inboundCodeCount}; ${trace(convertedResponse)}`, OPTIONS) - if (plans.length !== Number(summary.inboundCodeCount)) return false + && convertedInboundCodes.size === inboundCodes, + `plans=${plans.length}, inboundCodes=${convertedInboundCodes.size}, expected=${inboundCodes}; ${trace(convertedResponse)}`, OPTIONS) + if (plans.length !== inboundCodes) return false const replayResponse = await context.requestOverseas('POST', '/api/platform/v1/last-leg/inbound-plans/convert-container', body, { 'Idempotency-Key': key }, context.overseasToken) @@ -307,6 +315,15 @@ async function convertReceivingContainers(context, producers) { success(replayResponse) && JSON.stringify(replayResponse.body?.data) === JSON.stringify(plans), trace(replayResponse), OPTIONS) if (!success(replayResponse)) return false + const progressResponse = await context.requestOverseas('GET', + `/api/platform/v1/last-leg/orders/container-summaries/${encodeURIComponent(summary.sourceContainerBookingId)}`, + undefined, {}, context.overseasToken) + const progress = success(progressResponse) ? progressResponse.body?.data?.summary : null + context.addStep(MODULE, `整柜转换后计划完整度闭环 ${containerNo}`, progressResponse, + progress?.allPlansReady === true && Number(progress?.planCount) === totalOrders + && Number(progress?.pendingPlanCount) === 0, + `allPlansReady=${progress?.allPlansReady === true}, plans=${progress?.planCount ?? 'missing'}, pending=${progress?.pendingPlanCount ?? 'missing'}, plannedArrivalAt=${progress?.plannedArrivalAt || 'missing'}; ${trace(progressResponse)}`, OPTIONS) + if (progress?.allPlansReady !== true) return false } return true }