优化

2026-08-27 14:53:27 +08:00
parent ed3c1dafcc
commit d7bff980ca
18 changed files with 991 additions and 46 deletions
@@ -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;
+}
@@ -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<LoadingCompletionCommandEntity> {
+    @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<CURRENT_TIMESTAMP(3))
+         ORDER BY next_retry_at,id LIMIT #{limit}
+        """)
+    List<LoadingCompletionCommandEntity> 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<CURRENT_TIMESTAMP(3))
+        """)
+    int claim(@Param("id") long id, @Param("worker") String worker,
+              @Param("leaseUntil") LocalDateTime leaseUntil);
+
+    @Update("""
+        UPDATE fl_loading_completion_command SET lease_until=#{leaseUntil},updated_at=CURRENT_TIMESTAMP(3)
+         WHERE id=#{id} AND command_status='PROCESSING' AND lease_owner=#{worker}
+        """)
+    int renew(@Param("id") long id, @Param("worker") String worker,
+              @Param("leaseUntil") LocalDateTime leaseUntil);
+
+    @Update("""
+        UPDATE fl_loading_completion_command
+           SET command_status='COMPLETED',completed_at=CURRENT_TIMESTAMP(3),updated_at=CURRENT_TIMESTAMP(3),
+               lease_owner=NULL,lease_until=NULL,failure_stage=NULL,failure_code=NULL,
+               failure_request_id=NULL,failure_trace_id=NULL
+         WHERE id=#{id} AND command_status='PROCESSING' AND lease_owner=#{worker}
+        """)
+    int complete(@Param("id") long id, @Param("worker") String worker);
+
+    @Update("""
+        UPDATE fl_loading_completion_command
+           SET command_status=#{status},next_retry_at=#{nextRetryAt},updated_at=CURRENT_TIMESTAMP(3),
+               lease_owner=NULL,lease_until=NULL,failure_stage=#{stage},failure_code=#{code},
+               failure_request_id=#{requestId},failure_trace_id=#{traceId}
+         WHERE id=#{id} AND command_status='PROCESSING' AND lease_owner=#{worker}
+        """)
+    int fail(@Param("id") long id, @Param("worker") String worker, @Param("status") String status,
+             @Param("nextRetryAt") LocalDateTime nextRetryAt, @Param("stage") String stage,
+             @Param("code") String code, @Param("requestId") String requestId, @Param("traceId") String traceId);
+
+    @Update("""
+        UPDATE fl_loading_completion_command
+           SET command_status='RETRYABLE',next_retry_at=CURRENT_TIMESTAMP(3),updated_at=CURRENT_TIMESTAMP(3),
+               lease_owner=NULL,lease_until=NULL
+         WHERE id=#{id} AND command_status='FAILED'
+        """)
+    int resume(@Param("id") long id);
+
+    @Select("""
+        SELECT checkpoint_stage FROM fl_loading_completion_checkpoint
+         WHERE command_id=#{commandId} AND tenant_id=#{tenantId} AND inbound_id=#{inboundId}
+        """)
+    String checkpoint(@Param("commandId") long commandId, @Param("tenantId") long tenantId,
+                      @Param("inboundId") long inboundId);
+
+    @Insert("""
+        INSERT INTO fl_loading_completion_checkpoint
+          (command_id,tenant_id,inbound_id,owner_sequence,checkpoint_stage)
+        VALUES (#{commandId},#{tenantId},#{inboundId},#{sequence},#{stage})
+        ON DUPLICATE KEY UPDATE checkpoint_stage=VALUES(checkpoint_stage),updated_at=CURRENT_TIMESTAMP(3)
+        """)
+    int checkpointCompleted(@Param("commandId") long commandId, @Param("tenantId") long tenantId,
+                            @Param("inboundId") long inboundId, @Param("sequence") int sequence,
+                            @Param("stage") String stage);
+}
@@ -19,6 +19,29 @@ public interface LoadingPickTaskMapper extends BaseMapper<LoadingPickTaskEntity>
        + "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<LoadingPickTaskEntity>
               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<LoadingPickTaskEntity>
          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})
           <if test="keyword != null and keyword != ''">
             AND (booking.container_no LIKE CONCAT('%', #{keyword}, '%')
@@ -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<LoadingCompletionCommandEntity> 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);
+    }
+}
@@ -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<LoadingPickTaskEntity>()
            .eq(LoadingPickTaskEntity::getLoadingPlanId, planId));
@@ -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);
+    }
+}
@@ -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<LoadingPickItemMapper.InboundOwner> 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<LoadingPickItemMapper.InboundOwner> 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<String> 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());
@@ -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<LoadingPickItemEntity> taskItems = picks.items(taskId);
+        Map<Long, Integer> 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<LoadingPickItemEntity> 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<String> settlementTransferFacts(
        List<WarehousePalletLocationClient.SettlementTransferItem> remainingItems) {
        return remainingItems.stream().map(item -> item.reservationId() + ":" + item.remainingPackageCount())