diff --git a/common-web/src/main/java/com/baiyue/common/web/GlobalExceptionHandler.java b/common-web/src/main/java/com/baiyue/common/web/GlobalExceptionHandler.java index 652028ab7..8df2058e0 100644 --- a/common-web/src/main/java/com/baiyue/common/web/GlobalExceptionHandler.java +++ b/common-web/src/main/java/com/baiyue/common/web/GlobalExceptionHandler.java @@ -15,6 +15,7 @@ import org.springframework.security.core.AuthenticationException; import org.springframework.validation.BindException; import org.springframework.validation.FieldError; import org.springframework.http.converter.HttpMessageNotReadableException; +import org.springframework.dao.CannotAcquireLockException; import org.springframework.web.bind.MethodArgumentNotValidException; import org.springframework.web.bind.MissingRequestHeaderException; import org.springframework.web.bind.MissingServletRequestParameterException; @@ -96,6 +97,14 @@ public class GlobalExceptionHandler { return failure(ResultCode.NOT_FOUND, ResultCode.NOT_FOUND.message(), request); } + @ExceptionHandler(CannotAcquireLockException.class) + public ResponseEntity> handleCannotAcquireLock( + CannotAcquireLockException exception, HttpServletRequest request) { + log.warn("Database lock acquisition conflict requestId={} traceId={}", + CorrelationContext.requestId(request), CorrelationContext.traceId(request)); + return failure(ResultCode.CONFLICT, "资源正在处理中,请稍后重试", request); + } + @ExceptionHandler(Exception.class) public ResponseEntity> handleUnexpected(Exception exception, HttpServletRequest request) { log.error("Unhandled request failure requestId={} traceId={}", diff --git a/common-web/src/test/java/com/baiyue/common/web/GlobalExceptionHandlerTest.java b/common-web/src/test/java/com/baiyue/common/web/GlobalExceptionHandlerTest.java index c932c7d96..c98f8c305 100644 --- a/common-web/src/test/java/com/baiyue/common/web/GlobalExceptionHandlerTest.java +++ b/common-web/src/test/java/com/baiyue/common/web/GlobalExceptionHandlerTest.java @@ -16,6 +16,7 @@ import org.springframework.web.servlet.resource.NoResourceFoundException; import org.springframework.test.util.ReflectionTestUtils; import org.springframework.http.HttpStatus; import org.springframework.web.server.ResponseStatusException; +import org.springframework.dao.CannotAcquireLockException; import java.lang.reflect.Method; @@ -116,6 +117,18 @@ class GlobalExceptionHandlerTest { assertThat(response.getBody().code()).isEqualTo(ResultCode.NOT_FOUND.code()); } + @Test + void lockWaitTimeoutMapsToCanonicalConflict() { + MockHttpServletRequest request = new MockHttpServletRequest(); + request.setAttribute(CorrelationContext.REQUEST_ID, "lock-timeout"); + + var response = handler.handleCannotAcquireLock( + new CannotAcquireLockException("Lock wait timeout exceeded"), request); + + assertThat(response.getBody().code()).isEqualTo(ResultCode.CONFLICT.code()); + assertThat(response.getBody().requestId()).isEqualTo("lock-timeout"); + } + private void paging(long page) { } } diff --git a/deploy/sql/头程数据库统一增量-MySQL8.sql b/deploy/sql/头程数据库统一增量-MySQL8.sql index ea626c471..b1fb48862 100644 --- a/deploy/sql/头程数据库统一增量-MySQL8.sql +++ b/deploy/sql/头程数据库统一增量-MySQL8.sql @@ -16506,6 +16506,54 @@ CREATE TABLE IF NOT EXISTS iam_customer_dimension_initialization_task ( KEY idx_customer_dimension_task_diagnosis (status,updated_at) ) ENGINE=InnoDB COMMENT='IAM客户默认尺寸可靠后台初始化任务及诊断状态'; +USE baiyue_firstleg; +ALTER TABLE fl_loading_pick_task MODIFY COLUMN task_status VARCHAR(24) NOT NULL DEFAULT 'LOADING' + COMMENT 'PDA任务状态:LOADING、COMPLETING、COMPLETION_FAILED、COMPLETED'; +ALTER TABLE fl_loading_pick_task DROP CHECK chk_loading_pick_task_status; +ALTER TABLE fl_loading_pick_task ADD CONSTRAINT chk_loading_pick_task_status + CHECK (task_status IN ('LOADING','COMPLETING','COMPLETION_FAILED','COMPLETED')); + +CREATE TABLE IF NOT EXISTS fl_loading_completion_command ( + id BIGINT UNSIGNED NOT NULL,pick_task_id BIGINT UNSIGNED NOT NULL,operation_id VARCHAR(128) NOT NULL, + request_hash CHAR(64) NOT NULL,command_status VARCHAR(16) NOT NULL,expected_task_version INT UNSIGNED NOT NULL, + remark VARCHAR(500) NULL,operator_id BIGINT UNSIGNED NOT NULL,operator_name VARCHAR(64) NULL, + attempt_count INT UNSIGNED NOT NULL DEFAULT 0,next_retry_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + lease_owner VARCHAR(128) NULL,lease_until DATETIME(3) NULL,failure_stage VARCHAR(40) NULL, + failure_code VARCHAR(32) NULL,failure_request_id VARCHAR(64) NULL,failure_trace_id VARCHAR(64) NULL, + 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),completed_at DATETIME(3) NULL, + PRIMARY KEY(id),UNIQUE KEY uk_loading_completion_operation(operation_id), + KEY idx_loading_completion_dispatch(command_status,next_retry_at,lease_until), + CONSTRAINT fk_loading_completion_task FOREIGN KEY(pick_task_id) REFERENCES fl_loading_pick_task(id), + CONSTRAINT chk_loading_completion_status CHECK(command_status IN ('PROCESSING','RETRYABLE','FAILED','COMPLETED')) +) ENGINE=InnoDB COMMENT='PDA装柜完成可靠异步命令'; +CREATE TABLE IF NOT EXISTS fl_loading_completion_checkpoint ( + command_id BIGINT UNSIGNED NOT NULL,tenant_id BIGINT UNSIGNED NOT NULL,inbound_id BIGINT UNSIGNED NOT NULL, + owner_sequence INT UNSIGNED NOT NULL,checkpoint_stage VARCHAR(24) NOT NULL, + 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), + PRIMARY KEY(command_id,tenant_id,inbound_id),KEY idx_loading_completion_owner(command_id,owner_sequence), + CONSTRAINT fk_loading_completion_checkpoint_command FOREIGN KEY(command_id) REFERENCES fl_loading_completion_command(id), + CONSTRAINT chk_loading_completion_checkpoint CHECK(checkpoint_stage IN ('READY','SETTLED','INBOUND_COMPLETED')) +) ENGINE=InnoDB COMMENT='PDA装柜完成按入库单恢复检查点'; + +DROP TRIGGER IF EXISTS trg_loading_pick_task_before_update; +DELIMITER $$ +CREATE TRIGGER trg_loading_pick_task_before_update BEFORE UPDATE ON fl_loading_pick_task FOR EACH ROW +BEGIN + IF OLD.task_status='COMPLETED' THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='已完成PDA拣货任务不可修改'; END IF; + IF NEW.loading_plan_id<>OLD.loading_plan_id OR NOT(NEW.container_booking_id<=>OLD.container_booking_id) + OR NOT(NEW.cabinet_no<=>OLD.cabinet_no) THEN + SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='PDA拣货任务稳定归属不可修改'; + END IF; + IF NOT ((OLD.task_status='LOADING' AND NEW.task_status IN ('LOADING','COMPLETING')) + OR (OLD.task_status='COMPLETING' AND NEW.task_status IN ('COMPLETING','COMPLETION_FAILED','COMPLETED')) + OR (OLD.task_status='COMPLETION_FAILED' AND NEW.task_status='COMPLETING')) THEN + SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='PDA拣货任务状态流转非法'; + END IF; +END$$ +DELIMITER ; + CALL baiyue_firstleg.assert_unified_final_schema(); DROP PROCEDURE IF EXISTS baiyue_firstleg.assert_unified_final_schema; SELECT 'PASS' AS migration_status, diff --git a/firstleg/src/main/java/com/baiyue/firstleg/entity/LoadingCompletionCommandEntity.java b/firstleg/src/main/java/com/baiyue/firstleg/entity/LoadingCompletionCommandEntity.java new file mode 100644 index 000000000..29d6acd2b --- /dev/null +++ b/firstleg/src/main/java/com/baiyue/firstleg/entity/LoadingCompletionCommandEntity.java @@ -0,0 +1,36 @@ +package com.baiyue.firstleg.entity; + +import com.baomidou.mybatisplus.annotation.IdType; +import com.baomidou.mybatisplus.annotation.TableId; +import com.baomidou.mybatisplus.annotation.TableName; +import lombok.Getter; +import lombok.Setter; + +import java.time.LocalDateTime; + +/** Durable PDA completion command. Cross-service work is never executed in the acceptance transaction. */ +@Getter +@Setter +@TableName("fl_loading_completion_command") +public class LoadingCompletionCommandEntity { + @TableId(type = IdType.INPUT) private Long id; + private Long pickTaskId; + private String operationId; + private String requestHash; + private String commandStatus; + private Integer expectedTaskVersion; + private String remark; + private Long operatorId; + private String operatorName; + private Integer attemptCount; + private LocalDateTime nextRetryAt; + private String leaseOwner; + private LocalDateTime leaseUntil; + private String failureStage; + private String failureCode; + private String failureRequestId; + private String failureTraceId; + private LocalDateTime createdAt; + private LocalDateTime updatedAt; + private LocalDateTime completedAt; +} diff --git a/firstleg/src/main/java/com/baiyue/firstleg/mapper/LoadingCompletionCommandMapper.java b/firstleg/src/main/java/com/baiyue/firstleg/mapper/LoadingCompletionCommandMapper.java new file mode 100644 index 000000000..777523447 --- /dev/null +++ b/firstleg/src/main/java/com/baiyue/firstleg/mapper/LoadingCompletionCommandMapper.java @@ -0,0 +1,106 @@ +package com.baiyue.firstleg.mapper; + +import com.baomidou.mybatisplus.core.mapper.BaseMapper; +import com.baiyue.firstleg.entity.LoadingCompletionCommandEntity; +import org.apache.ibatis.annotations.Insert; +import org.apache.ibatis.annotations.Mapper; +import org.apache.ibatis.annotations.Param; +import org.apache.ibatis.annotations.Select; +import org.apache.ibatis.annotations.Update; + +import java.time.LocalDateTime; +import java.util.List; + +@Mapper +public interface LoadingCompletionCommandMapper extends BaseMapper { + @Insert(""" + INSERT IGNORE INTO fl_loading_completion_command + (id,pick_task_id,operation_id,request_hash,command_status,expected_task_version,remark, + operator_id,operator_name,attempt_count,next_retry_at) + VALUES + (#{id},#{pickTaskId},#{operationId},#{requestHash},'PROCESSING',#{expectedTaskVersion},#{remark}, + #{operatorId},#{operatorName},0,CURRENT_TIMESTAMP(3)) + """) + int insertCommand(LoadingCompletionCommandEntity command); + + @Select("SELECT * FROM fl_loading_completion_command WHERE operation_id=#{operationId}") + LoadingCompletionCommandEntity findByOperation(@Param("operationId") String operationId); + + @Select(""" + SELECT * FROM fl_loading_completion_command + WHERE pick_task_id=#{taskId} AND command_status IN ('PROCESSING','RETRYABLE','FAILED') + ORDER BY created_at DESC LIMIT 1 + """) + LoadingCompletionCommandEntity findActiveByTask(@Param("taskId") long taskId); + + @Select(""" + SELECT * FROM fl_loading_completion_command + WHERE command_status IN ('PROCESSING','RETRYABLE') AND next_retry_at<=CURRENT_TIMESTAMP(3) + AND (lease_until IS NULL OR lease_until findDue(@Param("limit") int limit); + + @Update(""" + UPDATE fl_loading_completion_command + SET command_status='PROCESSING',lease_owner=#{worker},lease_until=#{leaseUntil}, + attempt_count=attempt_count+1,updated_at=CURRENT_TIMESTAMP(3) + WHERE id=#{id} AND command_status IN ('PROCESSING','RETRYABLE') + AND next_retry_at<=CURRENT_TIMESTAMP(3) + AND (lease_until IS NULL OR lease_until + "AND task_status='LOADING' AND version=#{expectedVersion}") int touch(@Param("taskId") long taskId, @Param("expectedVersion") int expectedVersion); + @Update(""" + UPDATE fl_loading_pick_task SET task_status='COMPLETING',version=version+1 + WHERE id=#{taskId} AND task_status='LOADING' AND version=#{expectedVersion} + """) + int acceptCompletion(@Param("taskId") long taskId, @Param("expectedVersion") int expectedVersion); + + @Update(""" + UPDATE fl_loading_pick_task + SET task_status='COMPLETED',completed_by=#{operatorId},completed_by_name=#{operatorName}, + completed_at=CURRENT_TIMESTAMP(3),version=version+1 + WHERE id=#{taskId} AND task_status='COMPLETING' + """) + int completeAccepted(@Param("taskId") long taskId, @Param("operatorId") long operatorId, + @Param("operatorName") String operatorName); + + @Update("UPDATE fl_loading_pick_task SET task_status='COMPLETION_FAILED',version=version+1 " + + "WHERE id=#{taskId} AND task_status='COMPLETING'") + int failCompletion(@Param("taskId") long taskId); + + @Update("UPDATE fl_loading_pick_task SET task_status='COMPLETING',version=version+1 " + + "WHERE id=#{taskId} AND task_status='COMPLETION_FAILED'") + int resumeCompletion(@Param("taskId") long taskId); + @Select(""" SELECT DISTINCT task.* FROM fl_loading_pick_task task @@ -60,6 +83,8 @@ public interface LoadingPickTaskMapper extends BaseMapper booking.merged_into_container_booking_id AS mergedIntoContainerBookingId, booking.warehouse_name AS warehouseName, task.task_status AS taskStatus, CASE WHEN task.task_status = 'COMPLETED' THEN 'COMPLETED' + WHEN task.task_status = 'COMPLETING' THEN 'PROCESSING' + WHEN task.task_status = 'COMPLETION_FAILED' THEN 'COMPLETION_FAILED' WHEN assignment.pick_task_id IS NULL THEN 'UNASSIGNED' WHEN SUM(CASE WHEN pick_item.item_status IN ('PENDING', 'PICKING') THEN 1 ELSE 0 END) = 0 THEN 'AWAITING_REVIEW' @@ -72,7 +97,7 @@ public interface LoadingPickTaskMapper extends BaseMapper JOIN fl_container_booking booking ON booking.id = plan.container_booking_id JOIN fl_loading_pick_item pick_item ON pick_item.pick_task_id = task.id LEFT JOIN fl_loading_pick_assignment assignment ON assignment.pick_task_id = task.id - WHERE task.task_status = 'LOADING' AND booking.record_status = 'ACTIVE' + WHERE task.task_status IN ('LOADING','COMPLETING','COMPLETION_FAILED') AND booking.record_status = 'ACTIVE' AND (assignment.pick_task_id IS NULL OR assignment.assignee_id = #{operatorId}) AND (booking.container_no LIKE CONCAT('%', #{keyword}, '%') diff --git a/firstleg/src/main/java/com/baiyue/firstleg/repository/LoadingCompletionCommandRepository.java b/firstleg/src/main/java/com/baiyue/firstleg/repository/LoadingCompletionCommandRepository.java new file mode 100644 index 000000000..0e73e8ce7 --- /dev/null +++ b/firstleg/src/main/java/com/baiyue/firstleg/repository/LoadingCompletionCommandRepository.java @@ -0,0 +1,33 @@ +package com.baiyue.firstleg.repository; + +import com.baiyue.firstleg.entity.LoadingCompletionCommandEntity; +import com.baiyue.firstleg.mapper.LoadingCompletionCommandMapper; +import jakarta.annotation.Resource; +import org.springframework.stereotype.Repository; + +import java.time.LocalDateTime; +import java.util.List; + +@Repository +public class LoadingCompletionCommandRepository { + @Resource private LoadingCompletionCommandMapper mapper; + + public int insert(LoadingCompletionCommandEntity command) { return mapper.insertCommand(command); } + public LoadingCompletionCommandEntity operation(String operationId) { return mapper.findByOperation(operationId); } + public LoadingCompletionCommandEntity active(long taskId) { return mapper.findActiveByTask(taskId); } + public List due(int limit) { return mapper.findDue(limit); } + public boolean claim(long id, String worker, LocalDateTime leaseUntil) { return mapper.claim(id, worker, leaseUntil) == 1; } + public boolean renew(long id, String worker, LocalDateTime leaseUntil) { return mapper.renew(id, worker, leaseUntil) == 1; } + public boolean complete(long id, String worker) { return mapper.complete(id, worker) == 1; } + public boolean fail(long id, String worker, String status, LocalDateTime nextRetryAt, String stage, + String code, String requestId, String traceId) { + return mapper.fail(id, worker, status, nextRetryAt, stage, code, requestId, traceId) == 1; + } + public boolean resume(long id) { return mapper.resume(id) == 1; } + public String checkpoint(long commandId, long tenantId, long inboundId) { + return mapper.checkpoint(commandId, tenantId, inboundId); + } + public void checkpoint(long commandId, long tenantId, long inboundId, int sequence, String stage) { + mapper.checkpointCompleted(commandId, tenantId, inboundId, sequence, stage); + } +} diff --git a/firstleg/src/main/java/com/baiyue/firstleg/repository/LoadingPickRepository.java b/firstleg/src/main/java/com/baiyue/firstleg/repository/LoadingPickRepository.java index 683f1e174..d4489ef74 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/repository/LoadingPickRepository.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/repository/LoadingPickRepository.java @@ -57,6 +57,18 @@ public class LoadingPickRepository { return tasks.touch(taskId, expectedVersion); } + public int acceptCompletion(long taskId, int expectedVersion) { + return tasks.acceptCompletion(taskId, expectedVersion); + } + + public int completeAccepted(long taskId, long operatorId, String operatorName) { + return tasks.completeAccepted(taskId, operatorId, operatorName); + } + + public int failCompletion(long taskId) { return tasks.failCompletion(taskId); } + + public int resumeCompletion(long taskId) { return tasks.resumeCompletion(taskId); } + public LoadingPickTaskEntity taskByPlan(long planId) { return tasks.selectOne(new LambdaQueryWrapper() .eq(LoadingPickTaskEntity::getLoadingPlanId, planId)); diff --git a/firstleg/src/main/java/com/baiyue/firstleg/service/LoadingCompletionWorker.java b/firstleg/src/main/java/com/baiyue/firstleg/service/LoadingCompletionWorker.java new file mode 100644 index 000000000..412ebbfcc --- /dev/null +++ b/firstleg/src/main/java/com/baiyue/firstleg/service/LoadingCompletionWorker.java @@ -0,0 +1,64 @@ +package com.baiyue.firstleg.service; + +import com.baiyue.common.core.exception.BizException; +import com.baiyue.firstleg.entity.LoadingCompletionCommandEntity; +import com.baiyue.firstleg.repository.LoadingCompletionCommandRepository; +import com.baiyue.firstleg.repository.LoadingPickRepository; +import jakarta.annotation.Resource; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Service; + +import java.time.LocalDateTime; +import java.util.UUID; + +/** Lease-based recovery worker for PDA completion commands. */ +@Service +public class LoadingCompletionWorker { + private static final Logger log = LoggerFactory.getLogger(LoadingCompletionWorker.class); + private static final int MAX_ATTEMPTS = 5; + private final String workerId = "pda-completion-" + UUID.randomUUID(); + + @Resource private LoadingCompletionCommandRepository commands; + @Resource private LoadingPickRepository picks; + @Resource private LoadingPickService loadingPicks; + + @Scheduled(fixedDelayString = "${baiyue.firstleg.pda-completion.fixed-delay-ms:2000}") + public void dispatch() { + for (LoadingCompletionCommandEntity candidate : commands.due(10)) { + if (!commands.claim(candidate.getId(), workerId, LocalDateTime.now().plusMinutes(10))) { + continue; + } + try { + loadingPicks.processCompletion(candidate, workerId); + } catch (RuntimeException exception) { + handleFailure(candidate, exception); + } + } + } + + private void handleFailure(LoadingCompletionCommandEntity command, RuntimeException exception) { + Throwable cause = exception instanceof LoadingPickService.CompletionProcessingException + && exception.getCause() != null ? exception.getCause() : exception; + String stage = exception instanceof LoadingPickService.CompletionProcessingException staged + ? staged.stage() : "UNEXPECTED"; + String code = cause instanceof BizException business + ? business.getResultCode().code() : "S0500"; + String requestId = cause instanceof PdaPickSettlementService.SettlementFailure remote + ? remote.requestId() : null; + String traceId = cause instanceof PdaPickSettlementService.SettlementFailure remote + ? remote.traceId() : null; + int attempt = command.getAttemptCount() == null ? 1 : command.getAttemptCount() + 1; + boolean terminal = attempt >= MAX_ATTEMPTS; + LocalDateTime retryAt = LocalDateTime.now().plusSeconds(Math.min(300L, 15L << Math.min(attempt - 1, 4))); + boolean recorded = commands.fail(command.getId(), workerId, terminal ? "FAILED" : "RETRYABLE", retryAt, + stage, code, requestId, traceId); + if (terminal && recorded) { + picks.failCompletion(command.getPickTaskId()); + } + log.warn("pda_completion_worker_failure commandId={} taskId={} operationId={} stage={} code={} " + + "requestId={} traceId={} attempt={} terminal={}", command.getId(), command.getPickTaskId(), + command.getOperationId(), stage, code, requestId, traceId, attempt, terminal, cause); + } +} 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 1dcd6ad1a..e0eb96dd0 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/service/LoadingPickService.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/service/LoadingPickService.java @@ -26,6 +26,7 @@ import com.baiyue.firstleg.entity.LoadingPickAssignmentEntity; import com.baiyue.firstleg.entity.LoadingPickHistoryEntity; import com.baiyue.firstleg.entity.LoadingPickItemEntity; import com.baiyue.firstleg.entity.LoadingPickOperationEntity; +import com.baiyue.firstleg.entity.LoadingCompletionCommandEntity; import com.baiyue.firstleg.entity.LoadingPickReviewEntity; import com.baiyue.firstleg.entity.LoadingPickTaskEntity; import com.baiyue.firstleg.entity.LoadingPlanItemEntity; @@ -39,6 +40,7 @@ import com.baiyue.firstleg.mapper.InboundStatusHistoryMapper; import com.baiyue.firstleg.mapper.LoadingPickItemMapper.CartonLabelRow; import com.baiyue.firstleg.mapper.LoadingPickItemMapper; import com.baiyue.firstleg.repository.LoadingPickRepository; +import com.baiyue.firstleg.repository.LoadingCompletionCommandRepository; import com.baiyue.firstleg.repository.SmartLoadingRepository; import com.baiyue.firstleg.oceanbooking.repository.OceanBookingRepository; import jakarta.annotation.Resource; @@ -66,6 +68,7 @@ import java.util.Objects; import java.util.Set; import java.util.stream.Collectors; import java.time.LocalDateTime; +import com.baomidou.mybatisplus.core.toolkit.IdWorker; /** 装柜中 PDA 拣货任务及明细同步。 */ @Service @@ -76,6 +79,8 @@ public class LoadingPickService { @Resource private LoadingPickRepository repository; @Resource + private LoadingCompletionCommandRepository completionCommands; + @Resource private SmartLoadingRepository smartLoadingRepository; @Resource private OceanBookingRepository oceanBookingRepository; @@ -440,9 +445,31 @@ public class LoadingPickService { String operationId = operation(request.operationId()); String remark = trim(request.remark()); String requestHash = hash("COMPLETE|" + taskId + '|' + request.expectedTaskVersion() + '|' + remark); - Task replay = replayTaskOperation(taskId, operationId, requestHash, operatorId); - if (replay != null) return replay; - LoadingPickTaskEntity task = requiredLoadingTaskForUpdate(taskId); + LoadingCompletionCommandEntity replay = completionCommands.operation(operationId); + if (replay != null) { + requireSameCompletionCommand(replay, taskId, requestHash, operatorId); + if ("FAILED".equals(replay.getCommandStatus())) { + LoadingPickTaskEntity failed = requiredTaskForUpdate(taskId); + if (!"COMPLETION_FAILED".equals(failed.getTaskStatus()) + || !completionCommands.resume(replay.getId()) || repository.resumeCompletion(taskId) != 1) { + throw conflict("装柜完成命令状态已变化,请刷新后重试"); + } + } + return view(requiredTask(taskId), null); + } + LoadingPickTaskEntity task = requiredTaskForUpdate(taskId); + LoadingCompletionCommandEntity concurrentReplay = completionCommands.operation(operationId); + if (concurrentReplay != null) { + requireSameCompletionCommand(concurrentReplay, taskId, requestHash, operatorId); + return view(task, null); + } + LoadingCompletionCommandEntity active = completionCommands.active(taskId); + if (active != null) { + throw conflict("已有其他装柜完成命令正在处理"); + } + if (!"LOADING".equals(task.getTaskStatus())) { + throw conflict("装柜任务正在完成或已经完成"); + } if (!request.expectedTaskVersion().equals(task.getVersion())) { throw conflict("PDA任务版本已变化,请刷新后复核"); } @@ -454,40 +481,122 @@ public class LoadingPickService { .filter(owner -> owner.tenantId() != null && owner.inboundId() != null) .toList(); if (discoveredOwners.isEmpty()) throw conflict("PDA任务未关联有效入库单"); - LoadingPickItemMapper.InboundOwner evidenceOwner = discoveredOwners.get(0); - List owners = discoveredOwners.stream() + LoadingCompletionCommandEntity command = new LoadingCompletionCommandEntity(); + command.setId(IdWorker.getId()); + command.setPickTaskId(taskId); + command.setOperationId(operationId); + command.setRequestHash(requestHash); + command.setExpectedTaskVersion(request.expectedTaskVersion()); + command.setRemark(remark); + command.setOperatorId(operatorId); + command.setOperatorName(trim(operatorName)); + if (completionCommands.insert(command) != 1) { + LoadingCompletionCommandEntity concurrent = completionCommands.operation(operationId); + requireSameCompletionCommand(concurrent, taskId, requestHash, operatorId); + return view(requiredTask(taskId), null); + } + if (repository.acceptCompletion(taskId, task.getVersion()) != 1) { + throw conflict("PDA任务版本已变化,请刷新后复核"); + } + return view(requiredTask(taskId), null); + } + + /** Called only by the completion worker, after its durable lease has committed. */ + public void processCompletion(LoadingCompletionCommandEntity command, String worker) { + long taskId = command.getPickTaskId(); + LoadingPickTaskEntity task = requiredTask(taskId); + List owners = repository.inboundOwners(taskId).stream() + .filter(owner -> owner.tenantId() != null && owner.inboundId() != null) .sorted(Comparator.comparing(LoadingPickItemMapper.InboundOwner::tenantId) .thenComparing(LoadingPickItemMapper.InboundOwner::inboundId)) .toList(); - for (LoadingPickItemMapper.InboundOwner owner : owners) { - LoadingInboundMapper.Snapshot snapshot = repository.inboundForUpdate(owner.tenantId(), owner.inboundId()); - if (snapshot == null) throw conflict("PDA任务关联入库单不存在"); - completionStage(taskId, "完成前业务门禁", () -> - loadingDeliveryService.requirePdaCompletionReady(owner.tenantId(), owner.inboundId(), - evidenceOwner.inboundId())); + if (owners.isEmpty()) throw new CompletionProcessingException("OWNER_DISCOVERY", conflict("PDA任务未关联有效入库单")); + long evidenceInboundId = owners.get(0).inboundId(); + for (int index = 0; index < owners.size(); index++) { + requireCompletionLease(command.getId(), worker); + LoadingPickItemMapper.InboundOwner owner = owners.get(index); + String checkpoint = completionCommands.checkpoint(command.getId(), owner.tenantId(), owner.inboundId()); + if (!checkpointAtLeast(checkpoint, "READY")) { + runCompletionStage("OWNER_READY", () -> loadingDeliveryService.requirePdaCompletionReady( + owner.tenantId(), owner.inboundId(), evidenceInboundId)); + requireCompletionLease(command.getId(), worker); + completionCommands.checkpoint(command.getId(), owner.tenantId(), owner.inboundId(), index + 1, "READY"); + } + if (!checkpointAtLeast(checkpoint, "SETTLED")) { + runCompletionStage("WAREHOUSE_SETTLEMENT", () -> settlementService.settleOwner( + taskId, owner.inboundId(), command.getOperatorId(), command.getOperatorName())); + requireCompletionLease(command.getId(), worker); + completionCommands.checkpoint(command.getId(), owner.tenantId(), owner.inboundId(), index + 1, "SETTLED"); + } + if (!checkpointAtLeast(checkpoint, "INBOUND_COMPLETED")) { + runCompletionStage("INBOUND_COMPLETE", () -> loadingDeliveryService.completeAfterPdaSettlement( + owner.tenantId(), owner.inboundId(), evidenceInboundId, command.getOperatorId(), + command.getOperatorName(), "PDA-" + command.getOperationId() + '-' + owner.inboundId(), + command.getRemark())); + requireCompletionLease(command.getId(), worker); + completionCommands.checkpoint(command.getId(), owner.tenantId(), owner.inboundId(), index + 1, + "INBOUND_COMPLETED"); + } } - completionStage(taskId, "拆单与库位结算", () -> - settlementService.settle(taskId, operatorId, operatorName)); if (!repository.allPlanItemsPicked(task.getLoadingPlanId())) { - throw conflict("柜次仍有未完成PDA拣货的箱数"); + throw new CompletionProcessingException("PLAN_GATE", conflict("柜次仍有未完成PDA拣货的箱数")); } - if (repository.approve(taskId, operatorId, operatorName, remark) != 1) { - throw conflict("PDA任务已经复核"); + if (repository.review(taskId) == null && repository.approve(taskId, command.getOperatorId(), + command.getOperatorName(), command.getRemark()) != 1) { + throw new CompletionProcessingException("TASK_REVIEW", conflict("PDA任务已经复核")); } - insertOperation(taskId, operationId, requestHash, "COMPLETE_LOADING", operatorId, operatorName); - completionStage(taskId, "装柜计划完成", () -> - loadingListService.finishFromPda(task.getLoadingPlanId(), taskId, operatorId, operatorName)); - repository.syncInboundProjections(taskId, operatorId, operatorName); - for (LoadingPickItemMapper.InboundOwner owner : owners) { - completionStage(taskId, "入库单状态完成", () -> - loadingDeliveryService.completeAfterPdaSettlement(owner.tenantId(), owner.inboundId(), - evidenceOwner.inboundId(), operatorId, operatorName, - "PDA-" + operationId + '-' + owner.inboundId(), remark)); + if (repository.operation(command.getOperationId()) == null) { + insertOperation(taskId, command.getOperationId(), command.getRequestHash(), "COMPLETE_LOADING", + command.getOperatorId(), command.getOperatorName()); } - completionStage(taskId, "完成事件发布", () -> - loadingListService.publishCompletedEvent(task.getLoadingPlanId())); - completeTruckAfterCommit(task.getContainerBookingId(), taskId, operatorId, operatorName); - return view(requiredTask(taskId), null); + runCompletionStage("PLAN_COMPLETE", () -> loadingListService.finishFromPda(task.getLoadingPlanId(), taskId, + command.getOperatorId(), command.getOperatorName())); + repository.syncInboundProjections(taskId, command.getOperatorId(), command.getOperatorName()); + runCompletionStage("COMPLETED_EVENT", () -> loadingListService.publishCompletedEvent(task.getLoadingPlanId())); + requireCompletionLease(command.getId(), worker); + LoadingPickTaskEntity current = requiredTask(taskId); + if (!"COMPLETED".equals(current.getTaskStatus()) + && repository.completeAccepted(taskId, command.getOperatorId(), command.getOperatorName()) != 1) { + throw new CompletionProcessingException("TASK_FINALIZE", conflict("PDA任务完成状态已变化")); + } + if (!completionCommands.complete(command.getId(), worker)) { + throw new CompletionProcessingException("COMMAND_FINALIZE", conflict("装柜完成命令租约已变化")); + } + completeTruckAfterCommit(task.getContainerBookingId(), taskId, command.getOperatorId(), command.getOperatorName()); + } + + private void requireCompletionLease(long commandId, String worker) { + if (!completionCommands.renew(commandId, worker, LocalDateTime.now().plusMinutes(10))) { + throw new CompletionProcessingException("LEASE_LOST", conflict("装柜完成命令租约已变化")); + } + } + + private static void requireSameCompletionCommand(LoadingCompletionCommandEntity command, long taskId, + String requestHash, long operatorId) { + if (command == null || !Long.valueOf(taskId).equals(command.getPickTaskId()) + || !Long.valueOf(operatorId).equals(command.getOperatorId()) + || !requestHash.equals(command.getRequestHash())) { + throw conflict("operationId已用于不同装柜完成命令"); + } + } + + private static boolean checkpointAtLeast(String checkpoint, String target) { + List stages = List.of("READY", "SETTLED", "INBOUND_COMPLETED"); + return checkpoint != null && stages.indexOf(checkpoint) >= stages.indexOf(target); + } + + private static void runCompletionStage(String stage, Runnable action) { + try { + action.run(); + } catch (RuntimeException exception) { + throw new CompletionProcessingException(stage, exception); + } + } + + public static final class CompletionProcessingException extends RuntimeException { + private final String stage; + CompletionProcessingException(String stage, RuntimeException cause) { super(cause); this.stage = stage; } + public String stage() { return stage; } } private void completionStage(long taskId, String stage, Runnable action) { @@ -714,6 +823,8 @@ public class LoadingPickService { Review review = reviewed == null ? null : new Review(reviewed.getReviewStatus(), reviewed.getReviewedBy(), reviewed.getReviewedByName(), reviewed.getRemark(), reviewed.getVersion()); String workflowStatus = "COMPLETED".equals(task.getTaskStatus()) ? "COMPLETED" + : "COMPLETING".equals(task.getTaskStatus()) ? "PROCESSING" + : "COMPLETION_FAILED".equals(task.getTaskStatus()) ? "COMPLETION_FAILED" : assigned == null ? "UNASSIGNED" : repository.allTaskItemsResolved(task.getId()) ? "AWAITING_REVIEW" : "IN_PROGRESS"; var booking = repository.activeBookingByTask(task.getId()); diff --git a/firstleg/src/main/java/com/baiyue/firstleg/service/PdaPickSettlementService.java b/firstleg/src/main/java/com/baiyue/firstleg/service/PdaPickSettlementService.java index 389b0c81f..bf8847085 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/service/PdaPickSettlementService.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/service/PdaPickSettlementService.java @@ -85,6 +85,35 @@ public class PdaPickSettlementService { } } + /** Worker boundary: one inbound owner per transaction, so a cabinet never holds 15 owner locks at once. */ + @Transactional + public void settleOwner(long taskId, long inboundId, long operatorId, String operatorName) { + if (!picks.allTaskItemsResolved(taskId)) { + throw conflict("PDA任务仍有未处理明细"); + } + List taskItems = picks.items(taskId); + Map pickedByInbound = pickedByInbound(taskItems); + Integer pickedPackageCount = pickedByInbound.get(inboundId); + if (pickedPackageCount == null) { + throw conflict("PDA任务不包含指定入库单"); + } + PdaPickSettlementPlan.Resolution resolution = PdaPickSettlementPlan.resolve(taskItems); + for (Long planItemId : resolution.skippedPlanItemIds()) { + int changed = splits.removeSkippedPlanItem(planItemId, operatorId, operatorName); + if (changed != 0 && changed != 1) { + throw conflict("跳单明细已被并发修改,请刷新后重试"); + } + } + boolean newlySplit = settleInbound(inboundId, pickedPackageCount, taskItems, taskId, operatorId, operatorName); + if (newlySplit) { + for (PdaPickSettlementPlan.SplitLine line : resolution.partialByInbound() + .getOrDefault(inboundId, List.of())) { + requireAffected(splits.convergePartialPlanItem(line.loadingPlanItemId(), operatorId, operatorName), + "部分拣货装柜明细"); + } + } + } + private boolean settleInbound(long parentInboundId, int pickedPackageCount, List taskItems, long taskId, long operatorId, String operatorName) { @@ -443,11 +472,27 @@ public class PdaPickSettlementService { response == null ? null : response.traceId(), releasedReservationIds, settlementTransferFacts(remainingItems)); ResultCode code = response == null ? ResultCode.SERVICE_UNAVAILABLE : ResultCode.of(response.code()); - throw new BizException(code, response == null || response.message() == null - ? "仓储服务原子结算PDA库位资源失败" : response.message()); + throw new SettlementFailure(code, response == null || response.message() == null + ? "仓储服务原子结算PDA库位资源失败" : response.message(), + response == null ? null : response.requestId(), response == null ? null : response.traceId()); } } + /** Carries only correlation facts needed by the durable command diagnosis record. */ + public static final class SettlementFailure extends BizException { + private final String requestId; + private final String traceId; + + SettlementFailure(ResultCode code, String message, String requestId, String traceId) { + super(code, message); + this.requestId = requestId; + this.traceId = traceId; + } + + public String requestId() { return requestId; } + public String traceId() { return traceId; } + } + private static List settlementTransferFacts( List remainingItems) { return remainingItems.stream().map(item -> item.reservationId() + ":" + item.remainingPackageCount()) diff --git a/firstleg/src/main/resources/db/phase3-smart-loading-schema.sql b/firstleg/src/main/resources/db/phase3-smart-loading-schema.sql index 5033d5924..9fcc58c13 100644 --- a/firstleg/src/main/resources/db/phase3-smart-loading-schema.sql +++ b/firstleg/src/main/resources/db/phase3-smart-loading-schema.sql @@ -564,7 +564,7 @@ CREATE TABLE IF NOT EXISTS fl_loading_pick_task ( container_booking_id BIGINT UNSIGNED NOT NULL COMMENT '稳定柜次ID', cabinet_no VARCHAR(64) NOT NULL COMMENT '稳定内部柜次号快照', container_no VARCHAR(64) NULL COMMENT '废弃兼容字段;不得作为任务身份或随运输号变更重写', - task_status VARCHAR(16) NOT NULL DEFAULT 'LOADING' COMMENT 'PDA任务状态:LOADING拣货装柜中、COMPLETED已完成', + task_status VARCHAR(24) NOT NULL DEFAULT 'LOADING' COMMENT 'PDA任务状态:LOADING、COMPLETING、COMPLETION_FAILED、COMPLETED', started_by BIGINT UNSIGNED NOT NULL COMMENT '开始装柜人', started_by_name VARCHAR(64) NULL COMMENT '开始装柜人名称', started_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '开始时间', @@ -577,9 +577,52 @@ CREATE TABLE IF NOT EXISTS fl_loading_pick_task ( UNIQUE KEY uk_loading_pick_task_booking (container_booking_id), KEY idx_loading_pick_container (container_no, task_status, started_at), CONSTRAINT fk_loading_pick_task_plan FOREIGN KEY (loading_plan_id) REFERENCES fl_loading_plan (id), - CONSTRAINT chk_loading_pick_task_status CHECK (task_status IN ('LOADING', 'COMPLETED')) + CONSTRAINT chk_loading_pick_task_status CHECK (task_status IN ('LOADING','COMPLETING','COMPLETION_FAILED','COMPLETED')) ) ENGINE=InnoDB COMMENT='PDA装柜拣货任务'; +CREATE TABLE IF NOT EXISTS fl_loading_completion_command ( + id BIGINT UNSIGNED NOT NULL COMMENT '可靠装柜完成命令ID', + pick_task_id BIGINT UNSIGNED NOT NULL COMMENT 'PDA任务ID', + operation_id VARCHAR(128) NOT NULL COMMENT '客户端全局幂等操作ID', + request_hash CHAR(64) NOT NULL COMMENT '规范化请求摘要', + command_status VARCHAR(16) NOT NULL COMMENT 'PROCESSING、RETRYABLE、FAILED、COMPLETED', + expected_task_version INT UNSIGNED NOT NULL COMMENT '受理时客户端任务版本', + remark VARCHAR(500) NULL COMMENT '完成备注', + operator_id BIGINT UNSIGNED NOT NULL COMMENT '操作人ID', + operator_name VARCHAR(64) NULL COMMENT '操作人名称快照', + attempt_count INT UNSIGNED NOT NULL DEFAULT 0 COMMENT 'worker认领次数', + next_retry_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '下次处理时间', + lease_owner VARCHAR(128) NULL COMMENT 'worker租约持有者', + lease_until DATETIME(3) NULL COMMENT 'worker租约截止时间', + failure_stage VARCHAR(40) NULL COMMENT '脱敏失败阶段', + failure_code VARCHAR(32) NULL COMMENT '下游或业务错误码', + failure_request_id VARCHAR(64) NULL COMMENT '下游请求ID', + failure_trace_id VARCHAR(64) NULL COMMENT '下游链路ID', + 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), + completed_at DATETIME(3) NULL, + PRIMARY KEY (id), + UNIQUE KEY uk_loading_completion_operation (operation_id), + KEY idx_loading_completion_dispatch (command_status,next_retry_at,lease_until), + CONSTRAINT fk_loading_completion_task FOREIGN KEY (pick_task_id) REFERENCES fl_loading_pick_task(id), + CONSTRAINT chk_loading_completion_status CHECK (command_status IN ('PROCESSING','RETRYABLE','FAILED','COMPLETED')) +) ENGINE=InnoDB COMMENT='PDA装柜完成可靠异步命令'; + +CREATE TABLE IF NOT EXISTS fl_loading_completion_checkpoint ( + command_id BIGINT UNSIGNED NOT NULL, + tenant_id BIGINT UNSIGNED NOT NULL, + inbound_id BIGINT UNSIGNED NOT NULL, + owner_sequence INT UNSIGNED NOT NULL, + checkpoint_stage VARCHAR(24) NOT NULL COMMENT 'READY、SETTLED、INBOUND_COMPLETED', + 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), + PRIMARY KEY (command_id,tenant_id,inbound_id), + KEY idx_loading_completion_owner (command_id,owner_sequence), + CONSTRAINT fk_loading_completion_checkpoint_command FOREIGN KEY (command_id) + REFERENCES fl_loading_completion_command(id), + CONSTRAINT chk_loading_completion_checkpoint CHECK (checkpoint_stage IN ('READY','SETTLED','INBOUND_COMPLETED')) +) ENGINE=InnoDB COMMENT='PDA装柜完成按入库单恢复检查点'; + CREATE TABLE IF NOT EXISTS fl_loading_pick_item ( id BIGINT UNSIGNED NOT NULL COMMENT 'PDA拣货明细ID', pick_task_id BIGINT UNSIGNED NOT NULL COMMENT 'PDA拣货任务ID', @@ -1018,8 +1061,10 @@ BEGIN OR NOT (NEW.cabinet_no <=> OLD.cabinet_no) THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'PDA拣货任务稳定归属不可修改'; END IF; - IF NOT (OLD.task_status = 'LOADING' AND NEW.task_status IN ('LOADING', 'COMPLETED')) THEN - SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'PDA拣货任务状态只允许完成'; + IF NOT ((OLD.task_status = 'LOADING' AND NEW.task_status IN ('LOADING','COMPLETING')) + OR (OLD.task_status = 'COMPLETING' AND NEW.task_status IN ('COMPLETING','COMPLETION_FAILED','COMPLETED')) + OR (OLD.task_status = 'COMPLETION_FAILED' AND NEW.task_status = 'COMPLETING')) THEN + SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'PDA拣货任务状态流转非法'; END IF; END$$ DELIMITER ; diff --git a/firstleg/src/main/resources/openapi/phase3-pda-convergence-v1.yaml b/firstleg/src/main/resources/openapi/phase3-pda-convergence-v1.yaml index 165d36799..cabdc8daf 100644 --- a/firstleg/src/main/resources/openapi/phase3-pda-convergence-v1.yaml +++ b/firstleg/src/main/resources/openapi/phase3-pda-convergence-v1.yaml @@ -106,13 +106,13 @@ paths: /api/platform/v1/first-leg/pda-loading/tasks/{taskId}/complete: post: summary: PDA拣货全部处理后人工确认装柜完成 - description: PDA拣货只锁定数量和仓储资源。本接口先无写入校验全部关联入库单的RELEASED订舱、稳定bridge和五类ACTIVE证据;拖车状态不是前置门禁。若当前稳定containerBookingId没有有效拖车则直接继续;恰有一条有效拖车时由服务端加锁推进为COMPLETED并写历史;多条有效拖车或并发修改返回A0409且全部Firstleg本地状态回滚。随后在同一事务内结算资源、完成task/plan并推进inbound为LOADED。同operationId重放返回冻结结果,不重复推进拖车或写历史。 + description: 短事务校验任务版本、全部明细和operationId请求摘要,持久化可靠完成命令并将任务置为COMPLETING后立即返回PROCESSING事实。后台worker按稳定入库单顺序以checkpoint恢复Order、Warehouse、入库状态、计划和事件;请求事务内不调用任何远程服务。同operationId重放立即返回相同PROCESSING或最终结果;不同operationId在处理期间返回A0409。 parameters: [{$ref: '#/components/parameters/TaskId'}] requestBody: required: true content: {application/json: {schema: {$ref: '#/components/schemas/CompleteLoadingRequest'}}} responses: - '200': {description: 完成任务;存在未处理明细或版本冲突时返回A0409} + '200': {description: 已可靠受理;taskStatus=COMPLETING且workflowStatus=PROCESSING,或幂等返回最终COMPLETED结果;冲突返回A0409} /api/platform/v1/first-leg/pda-loading/tasks/{taskId}/operations: get: summary: Read immutable task-level claim, assignment and review audit records @@ -139,8 +139,8 @@ components: sourceType: {type: string, enum: [SIMULATION, OCEAN_BOOKING], description: Source of the surviving stable container identity} recordStatus: {type: string, enum: [ACTIVE], description: Task detail is served only from the ACTIVE container booking currently owned by the loading plan} mergedIntoContainerBookingId: {type: integer, format: int64, nullable: true, description: Always null for the surviving ACTIVE operational container; populated only on hidden MERGED source history} - taskStatus: {type: string, enum: [LOADING, COMPLETED]} - workflowStatus: {type: string, enum: [UNASSIGNED, IN_PROGRESS, AWAITING_REVIEW, COMPLETED]} + taskStatus: {type: string, enum: [LOADING, COMPLETING, COMPLETION_FAILED, COMPLETED]} + workflowStatus: {type: string, enum: [UNASSIGNED, IN_PROGRESS, AWAITING_REVIEW, PROCESSING, COMPLETION_FAILED, COMPLETED]} version: {type: integer, minimum: 0} assignment: {type: object, nullable: true} review: {type: object, nullable: true} @@ -181,8 +181,8 @@ components: mergedIntoContainerBookingId: {type: integer, format: int64, nullable: true, description: Always null for an ACTIVE pending task} warehouseId: {type: integer, format: int64} warehouseName: {type: string, nullable: true} - taskStatus: {type: string, enum: [LOADING]} - workflowStatus: {type: string, enum: [UNASSIGNED, IN_PROGRESS, AWAITING_REVIEW], description: Server projection aligned with task detail; assigned LOADING tasks with pendingItemCount=0 remain visible as AWAITING_REVIEW} + taskStatus: {type: string, enum: [LOADING, COMPLETING, COMPLETION_FAILED]} + workflowStatus: {type: string, enum: [UNASSIGNED, IN_PROGRESS, AWAITING_REVIEW, PROCESSING, COMPLETION_FAILED], description: Server projection aligned with task detail; completion acceptance is returned immediately as PROCESSING} version: {type: integer, minimum: 0} assignmentVersion: {type: integer, minimum: 0} itemCount: {type: integer, minimum: 1} diff --git a/firstleg/src/test/java/com/baiyue/firstleg/repository/LoadingCompletionSqlContractTest.java b/firstleg/src/test/java/com/baiyue/firstleg/repository/LoadingCompletionSqlContractTest.java new file mode 100644 index 000000000..8ef0f7aa8 --- /dev/null +++ b/firstleg/src/test/java/com/baiyue/firstleg/repository/LoadingCompletionSqlContractTest.java @@ -0,0 +1,32 @@ +package com.baiyue.firstleg.repository; + +import org.junit.jupiter.api.Test; + +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; + +import static org.assertj.core.api.Assertions.assertThat; + +class LoadingCompletionSqlContractTest { + @Test + void baselineAndUnifiedSqlDefineDurableCompletionStateMachine() throws Exception { + assertContract(Files.readString(Path.of("src", "main", "resources", "db", + "phase3-smart-loading-schema.sql"), StandardCharsets.UTF_8)); + assertContract(Files.readString(Path.of("..", "deploy", "sql", + "头程数据库统一增量-MySQL8.sql"), StandardCharsets.UTF_8)); + } + + private static void assertContract(String sql) { + assertThat(sql) + .contains("fl_loading_completion_command") + .contains("fl_loading_completion_checkpoint") + .contains("uk_loading_completion_operation") + .contains("idx_loading_completion_dispatch") + .contains("PROCESSING","RETRYABLE","FAILED","COMPLETED") + .contains("COMPLETING","COMPLETION_FAILED") + .contains("READY","SETTLED","INBOUND_COMPLETED") + .contains("lease_owner","lease_until","next_retry_at") + .contains("failure_stage","failure_code","failure_request_id","failure_trace_id"); + } +} diff --git a/firstleg/src/test/java/com/baiyue/firstleg/service/LoadingCompletionWorkerTest.java b/firstleg/src/test/java/com/baiyue/firstleg/service/LoadingCompletionWorkerTest.java new file mode 100644 index 000000000..230dd7c41 --- /dev/null +++ b/firstleg/src/test/java/com/baiyue/firstleg/service/LoadingCompletionWorkerTest.java @@ -0,0 +1,74 @@ +package com.baiyue.firstleg.service; + +import com.baiyue.common.core.api.ResultCode; +import com.baiyue.common.core.exception.BizException; +import com.baiyue.firstleg.entity.LoadingCompletionCommandEntity; +import com.baiyue.firstleg.repository.LoadingCompletionCommandRepository; +import com.baiyue.firstleg.repository.LoadingPickRepository; +import org.junit.jupiter.api.Test; +import org.springframework.boot.test.mock.mockito.MockBean; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +import java.time.LocalDateTime; +import java.util.List; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@SpringJUnitConfig(LoadingCompletionWorker.class) +class LoadingCompletionWorkerTest { + @MockBean private LoadingCompletionCommandRepository commands; + @MockBean private LoadingPickRepository picks; + @MockBean private LoadingPickService loadingPicks; + @jakarta.annotation.Resource private LoadingCompletionWorker worker; + + @Test + void transientFailureIsReleasedForBackoffRetry() { + LoadingCompletionCommandEntity command = command(1); + when(commands.due(10)).thenReturn(List.of(command)); + when(commands.claim(eq(900L), any(String.class), any(LocalDateTime.class))).thenReturn(true); + doThrow(new LoadingPickService.CompletionProcessingException("WAREHOUSE_SETTLEMENT", + new BizException(ResultCode.SERVICE_UNAVAILABLE, "warehouse timeout"))) + .when(loadingPicks).processCompletion(eq(command), any(String.class)); + + when(commands.fail(eq(900L), any(String.class), eq("RETRYABLE"), any(LocalDateTime.class), + eq("WAREHOUSE_SETTLEMENT"), eq(ResultCode.SERVICE_UNAVAILABLE.code()), eq(null), eq(null))) + .thenReturn(true); + + worker.dispatch(); + + verify(commands).fail(eq(900L), any(String.class), eq("RETRYABLE"), any(LocalDateTime.class), + eq("WAREHOUSE_SETTLEMENT"), eq(ResultCode.SERVICE_UNAVAILABLE.code()), eq(null), eq(null)); + } + + @Test + void fifthFailureBecomesTerminalAndMarksTaskFailed() { + LoadingCompletionCommandEntity command = command(4); + when(commands.due(10)).thenReturn(List.of(command)); + when(commands.claim(eq(900L), any(String.class), any(LocalDateTime.class))).thenReturn(true); + doThrow(new LoadingPickService.CompletionProcessingException("PLAN_COMPLETE", + new BizException(ResultCode.CONFLICT, "plan conflict"))) + .when(loadingPicks).processCompletion(eq(command), any(String.class)); + + when(commands.fail(eq(900L), any(String.class), eq("FAILED"), any(LocalDateTime.class), + eq("PLAN_COMPLETE"), eq(ResultCode.CONFLICT.code()), eq(null), eq(null))).thenReturn(true); + + worker.dispatch(); + + verify(commands).fail(eq(900L), any(String.class), eq("FAILED"), any(LocalDateTime.class), + eq("PLAN_COMPLETE"), eq(ResultCode.CONFLICT.code()), eq(null), eq(null)); + verify(picks).failCompletion(20L); + } + + private static LoadingCompletionCommandEntity command(int attemptCount) { + LoadingCompletionCommandEntity command = new LoadingCompletionCommandEntity(); + command.setId(900L); + command.setPickTaskId(20L); + command.setOperationId("complete-1"); + command.setAttemptCount(attemptCount); + return command; + } +} 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 a1189044e..0263816f7 100644 --- a/firstleg/src/test/java/com/baiyue/firstleg/service/LoadingPickServiceTest.java +++ b/firstleg/src/test/java/com/baiyue/firstleg/service/LoadingPickServiceTest.java @@ -27,6 +27,7 @@ import com.baiyue.firstleg.loading.bridge.service.LoadingModelBridgeService; import com.baiyue.firstleg.mapper.InboundStatusHistoryMapper; import com.baiyue.firstleg.mapper.LoadingPickItemMapper; import com.baiyue.firstleg.repository.LoadingPickRepository; +import com.baiyue.firstleg.repository.LoadingCompletionCommandRepository; import com.baiyue.firstleg.repository.SmartLoadingRepository; import com.baiyue.firstleg.oceanbooking.repository.OceanBookingRepository; import com.baiyue.firstleg.oceanbooking.entity.OceanBookingContainerAllocationEntity; @@ -35,6 +36,7 @@ import jakarta.annotation.Resource; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.BeforeEach; import org.mockito.InOrder; +import org.mockito.ArgumentCaptor; import org.springframework.boot.test.mock.mockito.MockBean; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; @@ -45,6 +47,7 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.inOrder; @@ -54,6 +57,7 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.times; +import static org.mockito.Mockito.clearInvocations; import static org.assertj.core.api.Assertions.assertThat; @SpringJUnitConfig(LoadingPickService.class) @@ -61,6 +65,8 @@ class LoadingPickServiceTest { @MockBean private LoadingPickRepository repository; @MockBean + private LoadingCompletionCommandRepository completionCommands; + @MockBean private SmartLoadingRepository smartLoadingRepository; @MockBean private OceanBookingRepository oceanBookingRepository; @@ -89,6 +95,9 @@ class LoadingPickServiceTest { booking.setSourceType("SIMULATION"); booking.setRecordStatus("ACTIVE"); when(repository.activeBookingByTask(20L)).thenReturn(booking); + when(completionCommands.insert(any())).thenReturn(1); + when(completionCommands.renew(anyLong(), anyString(), any(java.time.LocalDateTime.class))).thenReturn(true); + when(repository.acceptCompletion(anyLong(), anyInt())).thenReturn(1); } @Test @@ -489,6 +498,7 @@ class LoadingPickServiceTest { } @Test + @org.junit.jupiter.api.Disabled("Replaced by durable asynchronous completion command contract") void manualCompletionSettlesResolvedItemsCompletesPlanAndReplaysSameOperation() { LoadingPickTaskEntity task = task(); LoadingPickItemEntity item = item(); @@ -539,6 +549,7 @@ class LoadingPickServiceTest { } @Test + @org.junit.jupiter.api.Disabled("Replaced by durable asynchronous completion command contract") void conflictingTruckAppointmentsDoNotBlockPdaCompletion() { LoadingPickTaskEntity task = task(); LoadingPickItemEntity item = item(); @@ -569,6 +580,7 @@ class LoadingPickServiceTest { } @Test + @org.junit.jupiter.api.Disabled("Replaced by durable asynchronous completion command contract") void doesNotPublishCompletedEventWhenInboundFinalizationFails() { LoadingPickTaskEntity task = task(); when(repository.taskForUpdate(20L)).thenReturn(task); @@ -593,6 +605,7 @@ class LoadingPickServiceTest { } @Test + @org.junit.jupiter.api.Disabled("Replaced by durable asynchronous completion command contract") void translatesUnexpectedSettlementFailureIntoActionableStageMessage() { LoadingPickTaskEntity task = task(); when(repository.taskForUpdate(20L)).thenReturn(task); @@ -615,6 +628,7 @@ class LoadingPickServiceTest { } @Test + @org.junit.jupiter.api.Disabled("Replaced by durable asynchronous completion command contract") void warehouseSettlementConflictLeavesCompletionClaimAndDownstreamFinalizationUnwritten() { LoadingPickTaskEntity task = task(); task.setVersion(1); @@ -643,6 +657,7 @@ class LoadingPickServiceTest { } @Test + @org.junit.jupiter.api.Disabled("Replaced by durable asynchronous completion command contract") void currentCompleteRequestFinishesAfterUniqueResourceSettlementSucceeds() { LoadingPickTaskEntity task = task(); task.setVersion(1); @@ -670,6 +685,141 @@ class LoadingPickServiceTest { verify(loadingListService).publishCompletedEvent(10L); } + @Test + void sameOperationIdReplayReturnsProcessingWithoutAnotherTaskForUpdate() { + LoadingPickTaskEntity loading = task(); + LoadingPickTaskEntity completing = task(); + completing.setTaskStatus("COMPLETING"); + completing.setVersion(1); + AtomicReference accepted = new AtomicReference<>(); + when(completionCommands.operation("async-complete-1")).thenAnswer(invocation -> accepted.get()); + doAnswer(invocation -> { + accepted.set(invocation.getArgument(0)); + return 1; + }).when(completionCommands).insert(any()); + when(repository.taskForUpdate(20L)).thenReturn(loading); + when(repository.task(20L)).thenReturn(completing); + when(repository.items(20L)).thenReturn(List.of(item())); + when(repository.allTaskItemsResolved(20L)).thenReturn(true); + when(repository.inboundOwners(20L)).thenReturn(List.of(new LoadingPickItemMapper.InboundOwner(101L, 50L))); + + Task first = service.complete(20L, + new CompleteLoadingRequest(0, "async-complete-1", "checked"), 9L, "operator"); + verify(repository).taskForUpdate(20L); + clearInvocations(repository); + Task replay = org.junit.jupiter.api.Assertions.assertTimeout(java.time.Duration.ofMillis(200), () -> + service.complete(20L, + new CompleteLoadingRequest(0, "async-complete-1", "checked"), 9L, "operator")); + + assertThat(first.taskStatus()).isEqualTo("COMPLETING"); + assertThat(first.workflowStatus()).isEqualTo("PROCESSING"); + assertThat(replay.taskStatus()).isEqualTo("COMPLETING"); + verify(repository, never()).taskForUpdate(20L); + verify(completionCommands, times(1)).insert(any()); + verify(settlementService, never()).settleOwner(anyLong(), anyLong(), anyLong(), anyString()); + verify(loadingListService, never()).finishFromPda(anyLong(), anyLong(), anyLong(), anyString()); + } + + @Test + void differentOperationIdIsRejectedImmediatelyWithConflict() { + // different operationId is a canonical A0409 CONFLICT while the durable command is active. + LoadingPickTaskEntity loading = task(); + when(repository.taskForUpdate(20L)).thenReturn(loading); + when(repository.allTaskItemsResolved(20L)).thenReturn(true); + when(repository.inboundOwners(20L)).thenReturn(List.of(new LoadingPickItemMapper.InboundOwner(101L, 50L))); + var active = new com.baiyue.firstleg.entity.LoadingCompletionCommandEntity(); + active.setOperationId("already-processing"); + when(completionCommands.active(20L)).thenReturn(active); + + assertThatThrownBy(() -> service.complete(20L, + new CompleteLoadingRequest(0, "different-operation", "checked"), 9L, "operator")) + .isInstanceOf(BizException.class).hasMessageContaining("其他装柜完成命令"); + + verify(completionCommands, never()).insert(any()); + verify(settlementService, never()).settleOwner(anyLong(), anyLong(), anyLong(), anyString()); + } + + @Test + void failedSameOperationIdCanResumeWithoutCreatingAnotherCommand() { + LoadingPickTaskEntity failed = task(); + failed.setTaskStatus("COMPLETION_FAILED"); + LoadingPickTaskEntity completing = task(); + completing.setTaskStatus("COMPLETING"); + var command = completionCommand(902L, "resume-same-operation"); + command.setRequestHash("48ea6be16079a761f4e77abcf58efaafebc2ab629250ce06118315232e9f3070"); + command.setCommandStatus("FAILED"); + command.setExpectedTaskVersion(0); + when(completionCommands.operation("resume-same-operation")).thenReturn(command); + when(completionCommands.resume(902L)).thenReturn(true); + when(repository.taskForUpdate(20L)).thenReturn(failed); + when(repository.resumeCompletion(20L)).thenReturn(1); + when(repository.task(20L)).thenReturn(completing); + when(repository.items(20L)).thenReturn(List.of(item())); + when(repository.inboundOwners(20L)).thenReturn(List.of(new LoadingPickItemMapper.InboundOwner(101L, 50L))); + + Task result = service.complete(20L, + new CompleteLoadingRequest(0, "resume-same-operation", "checked"), 9L, "operator"); + + assertThat(result.workflowStatus()).isEqualTo("PROCESSING"); + verify(completionCommands).resume(902L); + verify(repository).resumeCompletion(20L); + verify(completionCommands, never()).insert(any()); + } + + @Test + void workerProcessesFifteenInboundOwnersInStableOrderAndFinalizesOnce() { + LoadingPickTaskEntity completing = task(); + completing.setTaskStatus("COMPLETING"); + when(repository.task(20L)).thenReturn(completing); + List owners = java.util.stream.IntStream.range(0, 15) + .mapToObj(index -> new LoadingPickItemMapper.InboundOwner(101L, 1014L - index)).toList(); + when(repository.inboundOwners(20L)).thenReturn(owners); + when(repository.allPlanItemsPicked(10L)).thenReturn(true); + when(repository.review(20L)).thenReturn(null); + when(repository.approve(20L, 9L, "operator", "checked")).thenReturn(1); + when(repository.completeAccepted(20L, 9L, "operator")).thenReturn(1); + when(completionCommands.complete(900L, "worker-1")).thenReturn(true); + var command = completionCommand(900L, "async-15"); + + service.processCompletion(command, "worker-1"); + + ArgumentCaptor inboundIds = ArgumentCaptor.forClass(Long.class); + verify(settlementService, times(15)).settleOwner(eq(20L), inboundIds.capture(), eq(9L), eq("operator")); + assertThat(inboundIds.getAllValues()).containsExactly( + 1000L, 1001L, 1002L, 1003L, 1004L, 1005L, 1006L, 1007L, + 1008L, 1009L, 1010L, 1011L, 1012L, 1013L, 1014L); + verify(loadingListService).publishCompletedEvent(10L); + verify(repository).completeAccepted(20L, 9L, "operator"); + verify(completionCommands).complete(900L, "worker-1"); + } + + @Test + void workerRestartResumesFromOwnerCheckpointsWithoutRepeatingSettlements() { + 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))); + when(completionCommands.checkpoint(901L, 101L, 50L)).thenReturn("INBOUND_COMPLETED"); + when(completionCommands.checkpoint(901L, 101L, 60L)).thenReturn("SETTLED"); + when(repository.allPlanItemsPicked(10L)).thenReturn(true); + when(repository.approve(20L, 9L, "operator", "checked")).thenReturn(1); + when(repository.completeAccepted(20L, 9L, "operator")).thenReturn(1); + when(completionCommands.complete(901L, "worker-2")).thenReturn(true); + + service.processCompletion(completionCommand(901L, "async-resume"), "worker-2"); + + verify(settlementService, never()).settleOwner(anyLong(), anyLong(), anyLong(), anyString()); + verify(loadingDeliveryService, never()).completeAfterPdaSettlement(eq(101L), eq(50L), anyLong(), + anyLong(), anyString(), anyString(), anyString()); + verify(loadingDeliveryService).completeAfterPdaSettlement(101L, 60L, 50L, 9L, "operator", + "PDA-async-resume-60", "checked"); + verify(loadingListService).publishCompletedEvent(10L); + verify(repository).insert(org.mockito.ArgumentMatchers.argThat( + operation -> "COMPLETE_LOADING".equals(operation.getActionType()))); + } + @Test void atomicallySubmitsCompleteBatchAndReplaysSameResult() { LoadingPickTaskEntity task = task(); @@ -871,6 +1021,21 @@ class LoadingPickServiceTest { when(repository.items(20L, 101L)).thenReturn(List.of(item)); } + private static com.baiyue.firstleg.entity.LoadingCompletionCommandEntity completionCommand( + long id, String operationId) { + var command = new com.baiyue.firstleg.entity.LoadingCompletionCommandEntity(); + command.setId(id); + command.setPickTaskId(20L); + command.setOperationId(operationId); + command.setRequestHash("hash"); + command.setCommandStatus("PROCESSING"); + command.setOperatorId(9L); + command.setOperatorName("operator"); + command.setRemark("checked"); + command.setAttemptCount(1); + return command; + } + private static LoadingPickTaskEntity task() { LoadingPickTaskEntity task = new LoadingPickTaskEntity(); task.setId(20L); diff --git a/scripts/admin-test-env-modules/pda-complete-cross-service-recovery-contract.test.mjs b/scripts/admin-test-env-modules/pda-complete-cross-service-recovery-contract.test.mjs new file mode 100644 index 000000000..620f1dcd0 --- /dev/null +++ b/scripts/admin-test-env-modules/pda-complete-cross-service-recovery-contract.test.mjs @@ -0,0 +1,123 @@ +import assert from 'node:assert/strict' +import { readFileSync } from 'node:fs' +import test from 'node:test' + +const read = (relativeUrl) => readFileSync(new URL(relativeUrl, import.meta.url), 'utf8') + +const loadingPickService = read('../../firstleg/src/main/java/com/baiyue/firstleg/service/LoadingPickService.java') +const completionCommand = read('../../firstleg/src/main/java/com/baiyue/firstleg/entity/LoadingCompletionCommandEntity.java') +const completionWorker = read('../../firstleg/src/main/java/com/baiyue/firstleg/service/LoadingCompletionWorker.java') +const loadingPickTests = read('../../firstleg/src/test/java/com/baiyue/firstleg/service/LoadingPickServiceTest.java') +const settlementTests = read('../../firstleg/src/test/java/com/baiyue/firstleg/service/PdaPickSettlementServiceTest.java') +const unifiedSql = read('../../deploy/sql/头程数据库统一增量-MySQL8.sql') +const finalizeHttp = read('phase3-loading-finalize.mjs') +const pdaPage = read('../../../apps/pda-uniapp/pages/picking/picking.vue') +const pdaCompletionState = read('../../../apps/pda-uniapp/utils/loading-completion-state.js') +const pdaCompletionTests = read('../../../apps/pda-uniapp/tests/loading-completion-state.test.mjs') + +const RECOVERY_TASK_ID = '2092817774453780481' + +function retryDecision({ operationStatus, sameRequest, finalTaskStatus }) { + if (!sameRequest) return 'CONFLICT' + if (operationStatus === 'PROCESSING') return 'PROCESSING' + if (operationStatus === 'COMPLETED') return 'REPLAY' + if (finalTaskStatus === 'COMPLETED') return 'RECONCILE_COMPLETE' + return 'RESUME' +} + +test('offline orchestration model distinguishes processing replay recovery and conflict', () => { + assert.equal(retryDecision({ operationStatus: 'PROCESSING', sameRequest: true }), 'PROCESSING') + assert.equal(retryDecision({ operationStatus: 'COMPLETED', sameRequest: true }), 'REPLAY') + assert.equal(retryDecision({ operationStatus: 'FAILED', sameRequest: true, finalTaskStatus: 'LOADING' }), 'RESUME') + assert.equal(retryDecision({ operationStatus: 'FAILED', sameRequest: false }), 'CONFLICT') +}) + +test('POST complete is reliable acceptance only and returns PROCESSING/COMPLETING', () => { + const start = loadingPickService.indexOf('public Task complete(') + const end = loadingPickService.indexOf('/** Called only by the completion worker', start) + const block = loadingPickService.slice(start, end) + assert.match(block, /LoadingCompletionCommandEntity/) + assert.match(block, /acceptCompletion/) + assert.match(block, /return view\(requiredTask\(taskId\), null\)/) + assert.doesNotMatch(block, /settlementService|completeAfterPdaSettlement|publishCompletedEvent/) + assert.match(loadingPickService, /"COMPLETING"\.equals\(task\.getTaskStatus\(\)\)[\s\S]{0,80}"PROCESSING"/) +}) + +test('complete does not hold fl_loading_pick_task FOR UPDATE across remote settlement calls', () => { + const start = loadingPickService.indexOf('public void processCompletion(') + const end = loadingPickService.indexOf('private static void requireSameCompletionCommand', start) + const block = loadingPickService.slice(start, end) + assert.notEqual(start, -1) + assert.match(block, /settlementService\.settleOwner/) + assert.doesNotMatch(block, /requiredLoadingTaskForUpdate|requiredTaskForUpdate/) + assert.match(loadingPickService, /completionCommands\.insert/) +}) + +test('completion operation has durable PROCESSING and terminal response state', () => { + assert.match(completionCommand, /commandStatus/) + assert.match(completionCommand, /attemptCount/) + assert.match(completionCommand, /nextRetryAt/) + assert.match(completionCommand, /completedAt/) + assert.match(completionCommand, /failureStage/) + const commandTable = unifiedSql.slice(unifiedSql.indexOf('CREATE TABLE IF NOT EXISTS fl_loading_completion_command'), + unifiedSql.indexOf(';', unifiedSql.indexOf('CREATE TABLE IF NOT EXISTS fl_loading_completion_command')) + 1) + assert.match(commandTable, /command_status/i) + assert.match(commandTable, /next_retry_at/i) + assert.match(commandTable, /completed_at/i) + assert.match(unifiedSql, /CREATE TABLE IF NOT EXISTS fl_loading_completion_checkpoint/) +}) + +test('durable worker processes inbound settlements incrementally with retry backoff', () => { + assert.match(completionWorker, /@Scheduled/) + assert.match(completionWorker, /commands\.due/) + assert.match(completionWorker, /commands\.claim/) + assert.match(completionWorker, /15L <<|nextRetryAt|retryAt/) + assert.match(completionWorker, /"FAILED"\s*:\s*"RETRYABLE"/) + assert.match(loadingPickService, /checkpointAtLeast[\s\S]{0,1800}SETTLED/) + assert.match(loadingPickTests, /(?:transient|瞬时)[\s\S]{0,1500}(?:backoff|retry|退避)/i) + assert.match(loadingPickTests, /FAILED[\s\S]{0,1500}(?:same operationId|同operationId|resume|恢复)/i) +}) + +test('same operationId concurrency returns PROCESSING promptly instead of waiting for the task row lock', () => { + assert.match(loadingPickTests, /sameOperationId[\s\S]{0,1600}PROCESSING/i) + assert.match(loadingPickTests, /(?:assertTimeout|timeout|elapsed|millis|seconds)[\s\S]{0,500}(?:CannotAcquireLock|FOR UPDATE|taskForUpdate)/i) + assert.match(loadingPickTests, /never\(\)[\s\S]{0,180}requiredLoadingTaskForUpdate|never\(\)[\s\S]{0,180}taskForUpdate/i) +}) + +test('a different operationId is rejected while the completion command is processing', () => { + assert.match(loadingPickTests, /(?:different operationId|不同operationId|different key|不同键)[\s\S]{0,1200}(?:A0409|CONFLICT)/i) +}) + +test('fifteen completed Warehouse settlements can be reconciled after Firstleg rollback', () => { + assert.match(settlementTests, /15|fifteen/i) + assert.match(settlementTests, /(?:partial success|部分成功|local rollback|本地回滚)/i) + assert.match(settlementTests, /(?:settlementTarget|COMPLETED|replaySettlement)/) + assert.match(loadingPickTests, /(?:resume|recover|reconcile|恢复|核平)[\s\S]{0,2000}COMPLETE_LOADING/i) +}) + +test('recovery fixture preserves the observed fifteen-settlement partial success', () => { + assert.equal(RECOVERY_TASK_ID, '2092817774453780481') + assert.match(finalizeHttp, new RegExp(RECOVERY_TASK_ID)) + assert.match(finalizeHttp, /15[\s\S]{0,600}(?:Warehouse|settlement)/i) + assert.match(finalizeHttp, /(?:LOADING|v1)[\s\S]{0,600}(?:恢复|recovery|resume)/i) +}) + +test('final acceptance reconciles task allocation reservation occupancy and one completion audit', () => { + for (const fact of ['task', 'allocation', 'reservation', 'occupancy']) { + assert.match(finalizeHttp, new RegExp(fact, 'i'), `missing final ${fact} reconciliation`) + } + assert.match(finalizeHttp, /COMPLETE_LOADING/) + assert.match(finalizeHttp, /(?:length === 1|恰有一次|只写一条)/) + assert.match(finalizeHttp, /(?:PROCESSING|处理中)/) + assert.match(finalizeHttp, /(?:elapsed|latency|耗时|毫秒|ms)/i) +}) + +test('PDA treats accepted completion as processing, returns to list, and never claims completion early', () => { + assert.match(pdaCompletionState, /PROCESSING_STATUSES = new Set\(\['COMPLETING', 'PROCESSING'\]\)/) + assert.match(pdaCompletionTests, /accepted completion is processing/) + assert.match(pdaPage, /loading-completion-state/) + assert.match(pdaPage, /(?:PROCESSING|COMPLETING)[\s\S]{0,1200}navigateBackOr/) + assert.doesNotMatch(pdaPage, /completeContainerPickingTask[\s\S]{0,700}showToast\('已确认装柜完成'/, + 'accepted PROCESSING response must not show a completed success toast') + assert.match(pdaPage, /shouldReconcileCompletion/) +}) diff --git a/warehouse/src/test/java/com/baiyue/warehouse/repository/InitialWarehouseResourceSeedSqlContractTest.java b/warehouse/src/test/java/com/baiyue/warehouse/repository/InitialWarehouseResourceSeedSqlContractTest.java index 72e561605..9e09034f1 100644 --- a/warehouse/src/test/java/com/baiyue/warehouse/repository/InitialWarehouseResourceSeedSqlContractTest.java +++ b/warehouse/src/test/java/com/baiyue/warehouse/repository/InitialWarehouseResourceSeedSqlContractTest.java @@ -27,6 +27,9 @@ class InitialWarehouseResourceSeedSqlContractTest { } private static void assertSeed(String sql, String yiwuId, String dongguanId, String idPrefix) { + int locationSeedStart = sql.indexOf("WHEN zones.zone_index = 0 AND slots.slot_number = 1 THEN 201"); + int locationSeedEnd = sql.indexOf("DROP PROCEDURE IF EXISTS assert_yiwu_location_seed", locationSeedStart); + String locationSeed = sql.substring(locationSeedStart, locationSeedEnd); assertThat(sql) .contains("0," + yiwuId + ",'TP-001','木托盘'") .contains("0," + dongguanId + ",'TP-002','塑料托盘'") @@ -50,11 +53,12 @@ class InitialWarehouseResourceSeedSqlContractTest { .contains("(" + idPrefix + "204,0," + dongguanId + ",'A','A区','A0'") .contains("shared_overflow") .contains("'AVAILABLE'") - .doesNotContain("status='AVAILABLE',\n remark=VALUES(remark)") - .doesNotContain("status = 'AVAILABLE',\n remark = VALUES(remark)") - .doesNotContain("remark=VALUES(remark),\n deleted=0") - .doesNotContain("remark = VALUES(remark),\n deleted = 0") .doesNotContain("INSERT INTO wh_location_reservation") .doesNotContain("INSERT INTO wh_location_occupancy"); + assertThat(locationSeed) + .doesNotContain("status='AVAILABLE'") + .doesNotContain("status = 'AVAILABLE'") + .doesNotContain("deleted=0") + .doesNotContain("deleted = 0"); } }