修复费用记录、装柜完成及海外履约资料

2026-10-21 23:20:09 +08:00
parent 5356a8c898
commit 2c3435adaf
62 changed files with 1622 additions and 298 deletions
@@ -4,6 +4,7 @@ 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.Options;
 import org.apache.ibatis.annotations.Param;
 import org.apache.ibatis.annotations.Select;
 import org.apache.ibatis.annotations.Update;
@@ -55,28 +56,34 @@ public interface LoadingCompletionCommandMapper extends BaseMapper<LoadingComple
    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}
+           AND lease_until>CURRENT_TIMESTAMP(3)
        """)
    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}
+           AND lease_until>CURRENT_TIMESTAMP(3)
        """)
    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}
+           AND lease_until>CURRENT_TIMESTAMP(3)
        """)
    int fail(@Param("id") long id, @Param("worker") String worker, @Param("status") String status,
             @Param("nextRetryAt") LocalDateTime nextRetryAt, @Param("stage") String stage,
@@ -97,13 +104,32 @@ public interface LoadingCompletionCommandMapper extends BaseMapper<LoadingComple
    String checkpoint(@Param("commandId") long commandId, @Param("tenantId") long tenantId,
                      @Param("inboundId") long inboundId);

+    /** 检查点事务锁住当前有效租约，与并发重新领取互斥。 */
+    @Options(useCache = false, flushCache = Options.FlushCachePolicy.TRUE)
+    @Select("""
+        SELECT id FROM fl_loading_completion_command
+         WHERE id=#{commandId} AND command_status='PROCESSING' AND lease_owner=#{worker}
+           AND lease_until>CURRENT_TIMESTAMP(3)
+         FOR UPDATE
+        """)
+    Long lockCheckpointLease(@Param("commandId") long commandId, @Param("worker") String worker);
+
+    /** 在已锁定的命令行上重验有效租约；检查点只能前进，不能被迟到写入降级。 */
    @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)
+        SELECT id,#{tenantId},#{inboundId},#{sequence},#{stage}
+          FROM fl_loading_completion_command
+         WHERE id=#{commandId} AND command_status='PROCESSING' AND lease_owner=#{worker}
+           AND lease_until>CURRENT_TIMESTAMP(3)
+        ON DUPLICATE KEY UPDATE checkpoint_stage=CASE
+          WHEN checkpoint_stage='INBOUND_COMPLETED' OR #{stage}='INBOUND_COMPLETED'
+            THEN 'INBOUND_COMPLETED'
+          WHEN checkpoint_stage='SETTLED' OR #{stage}='SETTLED' THEN 'SETTLED'
+          ELSE 'READY' END,updated_at=CURRENT_TIMESTAMP(3)
        """)
-    int checkpointCompleted(@Param("commandId") long commandId, @Param("tenantId") long tenantId,
+    int checkpointCompleted(@Param("commandId") long commandId, @Param("worker") String worker,
+                            @Param("tenantId") long tenantId,
                            @Param("inboundId") long inboundId, @Param("sequence") int sequence,
                            @Param("stage") String stage);
 }
@@ -11,16 +11,27 @@ import java.util.stream.Collectors;
@Service
 public class AppointmentForecastService {
    public record Forecast(Long containerBookingId,AppointmentForecastMapper.Fact latest,
-                           AppointmentForecastMapper.Fact previous,String timeBasis) { }
+                           AppointmentForecastMapper.Fact previous,String timeBasis,
+                           java.time.LocalDateTime portArrivalAt) { }
    private final AppointmentForecastMapper mapper;
-    public AppointmentForecastService(AppointmentForecastMapper mapper){this.mapper=mapper;}
+    private final com.baiyue.firstleg.terminaldrayage.repository.ContainerCalendarSourceRepository sources;
+    /** 读取船节点及日历权威到港来源，不从调用端接收推算日期。 */
+    public AppointmentForecastService(AppointmentForecastMapper mapper,
+        com.baiyue.firstleg.terminaldrayage.repository.ContainerCalendarSourceRepository sources) {
+        this.mapper=mapper; this.sources=sources;
+    }
+    /** 返回明确来源柜的到港事实，保留未知日期。 */
    public List<Forecast> batch(List<Long> ids) {
        if(ids==null || ids.isEmpty() || ids.size()>200 || ids.stream().anyMatch(id->id==null || id<=0)
            || new HashSet<>(ids).size()!=ids.size()) throw new ResponseStatusException(HttpStatus.BAD_REQUEST,"每批须提供1至200个不重复的柜次");
        var facts=mapper.latest(ids).stream().collect(Collectors.groupingBy(AppointmentForecastMapper.Fact::containerBookingId));
+        var arrivals=sources.byIds(ids).stream().collect(Collectors.toMap(
+            com.baiyue.firstleg.terminaldrayage.dto.ContainerCalendarSource::containerBookingId,
+            java.util.function.Function.identity()));
        return ids.stream().map(id->{
            var rows=facts.getOrDefault(id,List.of());
-            return new Forecast(id,rows.isEmpty()?null:rows.get(0),rows.size()<2?null:rows.get(1),"SOURCE_LOCAL_UNSPECIFIED");
+            return new Forecast(id,rows.isEmpty()?null:rows.get(0),rows.size()<2?null:rows.get(1),"SOURCE_LOCAL_UNSPECIFIED",
+                arrivals.containsKey(id)?arrivals.get(id).arrivalAt():null);
        }).toList();
    }
 }
@@ -4,6 +4,7 @@ import com.baiyue.firstleg.entity.LoadingCompletionCommandEntity;
 import com.baiyue.firstleg.mapper.LoadingCompletionCommandMapper;
 import jakarta.annotation.Resource;
 import org.springframework.stereotype.Repository;
+import org.springframework.transaction.annotation.Transactional;

 import java.time.LocalDateTime;
 import java.util.List;
@@ -12,6 +13,9 @@ import java.util.List;
 public class LoadingCompletionCommandRepository {
    @Resource private LoadingCompletionCommandMapper mapper;

+    /** 按事务提交后的命令标识读取唯一完成请求。 */
+    public LoadingCompletionCommandEntity command(long id) { return mapper.selectById(id); }
+
    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); }
@@ -28,7 +32,13 @@ public class LoadingCompletionCommandRepository {
    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);
+    /** 命令行锁覆盖租约检查与检查点写入，旧执行者不能在租约易主后覆盖新结果。 */
+    @Transactional
+    public boolean checkpoint(long commandId, String worker, long tenantId, long inboundId,
+                              int sequence, String stage) {
+        if (mapper.lockCheckpointLease(commandId, worker) == null) return false;
+        int written = mapper.checkpointCompleted(commandId, worker, tenantId, inboundId, sequence, stage);
+        // 同阶段重放可能被驱动计为零行；命令行仍被本事务锁住，重验租约区分幂等与过期拒绝。
+        return written > 0 || mapper.lockCheckpointLease(commandId, worker) != null;
    }
 }
@@ -9,37 +9,77 @@ import jakarta.annotation.Resource;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.core.task.TaskExecutor;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.event.TransactionalEventListener;

 import java.time.LocalDateTime;
 import java.util.UUID;
+import java.util.concurrent.RejectedExecutionException;

-/** Lease-based recovery worker for PDA completion commands. */
+/** 事务提交后立即处理 PDA 完成命令，并按持久化租约定时恢复。 */
@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;
+    @Resource(name = "applicationTaskExecutor") private TaskExecutor completionExecutor;

-    @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);
-            }
+    /** 唤醒只发生在受理事务提交后，异步执行不延长 HTTP 受理请求。 */
+    @TransactionalEventListener
+    public void accepted(LoadingPickService.CompletionAccepted accepted) {
+        try {
+            completionExecutor.execute(() -> dispatchAccepted(accepted));
+        } catch (RejectedExecutionException exception) {
+            // 执行器繁忙或关闭时保留持久化命令，仍由定时扫描领取，不能影响已提交回执。
+            log.warn("pda_completion_wakeup_deferred commandId={}", accepted.commandId(), exception);
        }
    }

-    private void handleFailure(LoadingCompletionCommandEntity command, RuntimeException exception) {
+    /** 唤醒与恢复共用原子领取；同一命令不会被两个执行入口重复结算。 */
+    private void dispatchAccepted(LoadingPickService.CompletionAccepted accepted) {
+        try {
+            LoadingCompletionCommandEntity command = commands.command(accepted.commandId());
+            if (command != null) process(command, "immediate", accepted.acceptedAtNanos());
+        } catch (RuntimeException exception) {
+            log.warn("pda_completion_wakeup_failed commandId={}", accepted.commandId(), exception);
+        }
+    }
+
+    /** 扫描到期或租约过期命令，覆盖唤醒丢失、进程重启及失败退避恢复。 */
+    @Scheduled(fixedDelayString = "${baiyue.firstleg.pda-completion.fixed-delay-ms:2000}")
+    public void dispatch() {
+        for (LoadingCompletionCommandEntity candidate : commands.due(10)) {
+            process(candidate, "scheduled", System.nanoTime());
+        }
+    }
+
+    /** 保留租约、检查点和失败恢复语义，记录领取及实际处理耗时。 */
+    private void process(LoadingCompletionCommandEntity command, String source, long queuedAt) {
+        // 每次领取使用独立所有者，避免旧执行线程在租约过期、同进程重新领取后继续写入。
+        String leaseOwner = "pda-completion-" + UUID.randomUUID();
+        if (!commands.claim(command.getId(), leaseOwner, LocalDateTime.now().plusMinutes(10))) return;
+        long started = System.nanoTime();
+        boolean succeeded = false;
+        log.info("pda_completion_started commandId={} taskId={} source={} wakeupMs={}",
+            command.getId(), command.getPickTaskId(), source, (started - queuedAt) / 1_000_000);
+        try {
+            loadingPicks.processCompletion(command, leaseOwner);
+            succeeded = true;
+        } catch (RuntimeException exception) {
+            handleFailure(command, leaseOwner, exception);
+        } finally {
+            log.info("pda_completion_finished commandId={} taskId={} source={} durationMs={} succeeded={}",
+                command.getId(), command.getPickTaskId(), source, (System.nanoTime() - started) / 1_000_000,
+                succeeded);
+        }
+    }
+
+    /** 确定性失败关闭任务，其余故障按原退避与最大次数恢复。 */
+    private void handleFailure(LoadingCompletionCommandEntity command, String leaseOwner, RuntimeException exception) {
        Throwable cause = exception instanceof LoadingPickService.CompletionProcessingException
            && exception.getCause() != null ? exception.getCause() : exception;
        String stage = exception instanceof LoadingPickService.CompletionProcessingException staged
@@ -53,7 +93,7 @@ public class LoadingCompletionWorker {
        int attempt = command.getAttemptCount() == null ? 1 : command.getAttemptCount() + 1;
        boolean terminal = isDeterministicLocalBusinessFailure(stage, cause) || 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,
+        boolean recorded = commands.fail(command.getId(), leaseOwner, terminal ? "FAILED" : "RETRYABLE", retryAt,
            stage, code, requestId, traceId);
        if (terminal && recorded) {
            picks.failCompletion(command.getPickTaskId());
@@ -63,6 +103,7 @@ public class LoadingCompletionWorker {
            command.getOperationId(), stage, code, requestId, traceId, attempt, terminal, cause);
    }

+    /** 仅将已明确不可重试的本地业务门禁判为终态。 */
    private static boolean isDeterministicLocalBusinessFailure(String stage, Throwable cause) {
        if (!(cause instanceof BizException business)) {
            return false;
@@ -56,6 +56,7 @@ import jakarta.annotation.Resource;
 import org.springframework.dao.DuplicateKeyException;
 import org.springframework.stereotype.Service;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.transaction.support.TransactionTemplate;
 import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.StringUtils;
@@ -87,6 +88,8 @@ public class LoadingPickService {
    @Resource
    private LoadingCompletionCommandRepository completionCommands;
    @Resource
+    private ApplicationEventPublisher completionEvents;
+    @Resource
    private TransactionTemplate completionTransaction;
    @Resource
    private SmartLoadingRepository smartLoadingRepository;
@@ -552,6 +555,7 @@ public class LoadingPickService {
                    || !completionCommands.resume(replay.getId()) || repository.resumeCompletion(taskId) != 1) {
                    throw conflict("装柜完成命令状态已变化，请刷新后重试");
                }
+                completionEvents.publishEvent(new CompletionAccepted(replay.getId(), System.nanoTime()));
            }
            return view(requiredTask(taskId), null);
        }
@@ -598,9 +602,14 @@ public class LoadingPickService {
        if (repository.acceptCompletion(taskId, task.getVersion()) != 1) {
            throw conflict("PDA任务版本已变化，请刷新后复核");
        }
+        // 提交后立即唤醒仍经持久化租约领取；回滚不得触发结算，定时扫描继续负责故障恢复。
+        completionEvents.publishEvent(new CompletionAccepted(command.getId(), System.nanoTime()));
        return view(requiredTask(taskId), null);
    }

+    /** 本进程事务提交后的唤醒信号，权威命令和恢复事实仍保存在数据库。 */
+    public record CompletionAccepted(long commandId, long acceptedAtNanos) {}
+
    private void requireCompletionContainerNumber(long taskId) {
        ContainerBookingEntity booking = repository.activeBookingByTask(taskId);
        if (booking == null || !StringUtils.hasText(booking.getContainerNo())) {
@@ -613,7 +622,7 @@ public class LoadingPickService {
        }
    }

-    /** Called only by the completion worker, after its durable lease has committed. */
+    /** 仅在持久化租约领取成功后执行，远程结算不占用受理事务。 */
    public void processCompletion(LoadingCompletionCommandEntity command, String worker) {
        long taskId = command.getPickTaskId();
        inboundHoldGuard.requireTaskNotHeld(taskId);
@@ -625,55 +634,55 @@ public class LoadingPickService {
            .toList();
        if (owners.isEmpty()) throw new CompletionProcessingException("OWNER_DISCOVERY", conflict("PDA任务未关联有效入库单"));
        List<GroupSnapshot> taskGroups = coloadGroups(owners);
-        runCompletionStage("CABINET_EVIDENCE", () ->
+        runCompletionStage(taskId, "CABINET_EVIDENCE", () ->
            cabinetEvidenceService.requireComplete(task.getContainerBookingId()));
-        runCompletionStage("COMPLETED_EVENT_READY", () ->
+        runCompletionStage(taskId, "COMPLETED_EVENT_READY", () ->
            loadingListService.requireCompletedEventReady(task.getLoadingPlanId()));

-        // Phase 1 is read-only: cabinet evidence is checked once, then every owner must pass its business gates
-        // before any warehouse resource or inbound status can be changed. This prevents a later customer gate from
-        // leaving earlier customers partially completed.
+        // 全柜所有客户先通过只读门禁，再开始资源结算，避免后续客户失败造成前序客户局部完成。
        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, "SETTLED")) {
-                runCompletionStage("OWNER_READY", () -> loadingDeliveryService.requirePdaCompletionReady(
+                runCompletionStage(taskId, "OWNER_READY", () -> loadingDeliveryService.requirePdaCompletionReady(
                    owner.tenantId(), owner.inboundId()));
                if (!checkpointAtLeast(checkpoint, "READY")) {
-                    requireCompletionLease(command.getId(), worker);
-                    completionCommands.checkpoint(command.getId(), owner.tenantId(), owner.inboundId(),
+                    recordCompletionCheckpoint(command.getId(), worker, owner.tenantId(), owner.inboundId(),
                        index + 1, "READY");
                }
            }
        }

-        // Phase 2 is recoverable by per-owner checkpoints and stable idempotency keys. It starts only after the
-        // complete cabinet passed phase 1, while prior deployments' completed checkpoints remain safely resumable.
+        // 全柜门禁通过后，按客户检查点和稳定幂等键逐步结算，已完成的检查点可安全续做。
        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, "SETTLED")) {
-                runCompletionStage("WAREHOUSE_SETTLEMENT", () -> settlementService.settleOwner(
+                runCompletionStage(taskId, "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");
+                recordCompletionCheckpoint(command.getId(), worker, owner.tenantId(), owner.inboundId(), index + 1, "SETTLED");
            }
            if (!checkpointAtLeast(checkpoint, "INBOUND_COMPLETED")) {
-                runCompletionStage("INBOUND_COMPLETE", () -> loadingDeliveryService.completeAfterPdaSettlement(
+                runCompletionStage(taskId, "INBOUND_COMPLETE", () -> loadingDeliveryService.completeAfterPdaSettlement(
                    owner.tenantId(), owner.inboundId(),
                    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,
+                recordCompletionCheckpoint(command.getId(), worker, owner.tenantId(), owner.inboundId(), index + 1,
                    "INBOUND_COMPLETED");
            }
        }
-        completionTransaction.executeWithoutResult(ignored -> finalizeCompletion(command, worker, task, taskGroups));
+        long transactionStarted = System.nanoTime();
+        try {
+            completionTransaction.executeWithoutResult(ignored -> finalizeCompletion(command, worker, task, taskGroups));
+        } finally {
+            log.info("pda_completion_transaction taskId={} durationMs={}", taskId,
+                (System.nanoTime() - transactionStarted) / 1_000_000);
+        }
    }

-    /** Final Firstleg convergence only; all remote owner settlement calls have completed before this short transaction. */
+    /** 远程结算完成后，仅在短事务内收敛 Firstleg 本地完成事实。 */
    private void finalizeCompletion(LoadingCompletionCommandEntity command, String worker,
                                    LoadingPickTaskEntity task, List<GroupSnapshot> taskGroups) {
        long taskId = task.getId();
@@ -688,11 +697,11 @@ public class LoadingPickService {
            insertOperation(taskId, command.getOperationId(), command.getRequestHash(), "COMPLETE_LOADING",
                command.getOperatorId(), command.getOperatorName());
        }
-        runCompletionStage("PLAN_COMPLETE", () -> loadingListService.finishFromPda(task.getLoadingPlanId(), taskId,
+        runCompletionStage(taskId, "PLAN_COMPLETE", () -> loadingListService.finishFromPda(task.getLoadingPlanId(), taskId,
            command.getOperatorId(), command.getOperatorName()));
-        runCompletionStage("COMPLETED_EVENT", () -> loadingListService.publishCompletedEvent(task.getLoadingPlanId()));
+        runCompletionStage(taskId, "COMPLETED_EVENT", () -> loadingListService.publishCompletedEvent(task.getLoadingPlanId()));
        for (GroupSnapshot group : taskGroups) {
-            runCompletionStage("COLOAD_CONSUME", () -> coloadGroups.consumeWholeGroup(group.groupId(),
+            runCompletionStage(taskId, "COLOAD_CONSUME", () -> coloadGroups.consumeWholeGroup(group.groupId(),
                task.getLoadingPlanId(), "PDA-CONSUME-" + hash(command.getOperationId() + '|'
                    + group.groupId()).substring(0, 48), command.getOperatorId(), command.getOperatorName()));
        }
@@ -750,6 +759,14 @@ public class LoadingPickService {
        }
    }

+    /** 检查点失去当前有效租约时立即停止，不再推进后续客户或柜次。 */
+    private void recordCompletionCheckpoint(long commandId, String worker, long tenantId, long inboundId,
+                                            int sequence, String stage) {
+        if (!completionCommands.checkpoint(commandId, worker, tenantId, inboundId, sequence, stage)) {
+            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())
@@ -764,11 +781,18 @@ public class LoadingPickService {
        return checkpoint != null && stages.indexOf(checkpoint) >= stages.indexOf(target);
    }

-    private static void runCompletionStage(String stage, Runnable action) {
+    /** 记录实际阶段耗时，区分业务处理与调度、前端刷新等待。 */
+    private static void runCompletionStage(long taskId, String stage, Runnable action) {
+        long started = System.nanoTime();
+        boolean succeeded = false;
        try {
            action.run();
+            succeeded = true;
        } catch (RuntimeException exception) {
            throw new CompletionProcessingException(stage, exception);
+        } finally {
+            log.info("pda_completion_stage taskId={} stage={} durationMs={} succeeded={}",
+                taskId, stage, (System.nanoTime() - started) / 1_000_000, succeeded);
        }
    }

@@ -9,6 +9,25 @@ import java.util.List;

@Mapper
 public interface ContainerCalendarSourceMapper {
+    /** 按明确来源柜读取与预约日历相同的到港事实，实际到港优先。 */
+    @Select("""
+        <script>
+        SELECT c.id containerBookingId,c.cabinet_no cabinetNo,c.container_no containerNo,
+          c.vessel_name vesselName,c.voyage_no voyageNo,c.destination_port_name destinationPortName,
+          c.destination_warehouse_code destinationWarehouseCode,
+          COALESCE(n.actual_at,c.actual_port_arrival_at,c.estimated_port_arrival_at) arrivalAt,
+          CASE WHEN COALESCE(n.actual_at,c.actual_port_arrival_at) IS NOT NULL THEN 'ACTUAL' ELSE 'ESTIMATED' END arrivalType
+        FROM fl_container_booking c
+        LEFT JOIN fl_carrier_schedule_node_fact n ON n.id=(
+          SELECT f.id FROM fl_carrier_schedule_node_fact f WHERE f.container_booking_id=c.id
+            AND f.standard_node='ARRIVED_DESTINATION_PORT' AND f.actual_at IS NOT NULL
+          ORDER BY f.observed_at DESC,f.id DESC LIMIT 1)
+        WHERE c.record_status='ACTIVE' AND c.id IN
+        <foreach collection='ids' item='id' open='(' separator=',' close=')'>#{id}</foreach>
+        </script>
+        """)
+    List<ContainerCalendarSource> byIds(@Param("ids") List<Long> ids);
+
    /** 同时选出到港、缺日期和可能还柜的有效出仓柜，最终日期由业务服务复核。 */
    @Select("""
        SELECT c.id containerBookingId,c.cabinet_no cabinetNo,c.container_no containerNo,
@@ -11,6 +11,8 @@ public class ContainerCalendarSourceRepository {
    private final ContainerCalendarSourceMapper mapper;
    /** 使用本服务 Mapper 读取日历所需柜级来源。 */
    public ContainerCalendarSourceRepository(ContainerCalendarSourceMapper mapper) { this.mapper = mapper; }
+    /** 内部批量读取明确柜次，日期口径与平台日历一致。 */
+    public List<ContainerCalendarSource> byIds(List<Long> ids) { return mapper.byIds(ids); }
    /** 到港前移两天覆盖默认到仓，直送还柜候选额外留一天覆盖夏令时。 */
    public List<ContainerCalendarSource> sources(LocalDate from, LocalDate to) {
        return mapper.sources(from.atStartOfDay(), from.minusDays(2).atStartOfDay(),
@@ -34,6 +34,7 @@ paths:
                      properties:
                        containerBookingId: {type: string}
                        timeBasis: {type: string, enum: [SOURCE_LOCAL_UNSPECIFIED]}
+                        portArrivalAt: {type: string, format: date-time, nullable: true, description: 与预约日历相同的权威到港当地时间，最新有效实际到港优先，其次柜资料实际到港及预计到港；未知保持空}
                        latest: {$ref: '#/components/schemas/Fact'}
                        previous: {$ref: '#/components/schemas/Fact'}
        '401': {description: 内部身份缺失、错误或服务端未配置}
@@ -9,8 +9,20 @@ import static org.assertj.core.api.Assertions.*;
 import static org.mockito.Mockito.*;

 class AppointmentForecastServiceTest {
+    /** 无船期预计观察时仍返回日历所用的柜级到港事实，供海外统一推算日期。 */
+    @Test void returnsTheCalendarArrivalEvenWhenLatestForecastIsAbsent() {
+        var mapper=mock(AppointmentForecastMapper.class);
+        var sources=mock(com.baiyue.firstleg.terminaldrayage.repository.ContainerCalendarSourceRepository.class);
+        var arrival=LocalDateTime.of(2026,10,30,23,30);
+        when(sources.byIds(List.of(1L))).thenReturn(List.of(
+            new com.baiyue.firstleg.terminaldrayage.dto.ContainerCalendarSource(1L,"D0001","CONT1",
+                null,null,null,null,arrival,"ACTUAL")));
+        var result=new AppointmentForecastService(mapper,sources).batch(List.of(1L));
+        assertThat(result.get(0).latest()).isNull();
+        assertThat(result.get(0).portArrivalAt()).isEqualTo(arrival);
+    }
    @Test void preservesMissingForecastAndSourceLocalTimesWithoutSynthesizingEta() {
-        var mapper=mock(AppointmentForecastMapper.class);var service=new AppointmentForecastService(mapper);
+        var mapper=mock(AppointmentForecastMapper.class);var service=new AppointmentForecastService(mapper,mock(com.baiyue.firstleg.terminaldrayage.repository.ContainerCalendarSourceRepository.class));
        var observed=LocalDateTime.of(2026,10,10,12,0);
        var latest=new AppointmentForecastMapper.Fact(1L,20L,"carrier","REF",observed,null,null);
        var previous=new AppointmentForecastMapper.Fact(1L,10L,"carrier","REF-OLD",observed.minusDays(1),observed.plusDays(5),null);
@@ -21,7 +33,7 @@ class AppointmentForecastServiceTest {
        verify(mapper).latest(List.of(1L,2L));verifyNoMoreInteractions(mapper);
    }
    @Test void rejectsInvalidBatchesAndMissingOrBlankInternalCredentials() {
-        var mapper=mock(AppointmentForecastMapper.class);var service=new AppointmentForecastService(mapper);
+        var mapper=mock(AppointmentForecastMapper.class);var service=new AppointmentForecastService(mapper,mock(com.baiyue.firstleg.terminaldrayage.repository.ContainerCalendarSourceRepository.class));
        for(var ids:List.of(List.<Long>of(),List.of(1L,1L),List.of(-1L),Collections.nCopies(201,1L)))
            assertThatThrownBy(()->service.batch(ids)).isInstanceOf(org.springframework.web.server.ResponseStatusException.class);
        assertThatThrownBy(()->service.batch(null)).isInstanceOf(org.springframework.web.server.ResponseStatusException.class);
@@ -0,0 +1,190 @@
+package com.baiyue.firstleg.repository;
+
+import com.baomidou.mybatisplus.core.MybatisConfiguration;
+import com.baomidou.mybatisplus.extension.spring.MybatisSqlSessionFactoryBean;
+import com.baiyue.firstleg.mapper.LoadingCompletionCommandMapper;
+import org.h2.jdbcx.JdbcDataSource;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mybatis.spring.SqlSessionTemplate;
+import org.springframework.aop.framework.ProxyFactory;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.transaction.annotation.AnnotationTransactionAttributeSource;
+import org.springframework.transaction.interceptor.TransactionInterceptor;
+
+import java.sql.SQLException;
+import java.time.LocalDateTime;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.AdditionalAnswers.delegatesTo;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+
+/** 用真实 Mapper SQL、事务代理和独立连接验证完成检查点的租约隔离。 */
+class LoadingCompletionCheckpointLeaseTest {
+    private JdbcDataSource dataSource;
+    private JdbcTemplate jdbc;
+    private LoadingCompletionCommandMapper mapper;
+    private LoadingCompletionCommandRepository commands;
+
+    /** 每个用例使用独立内存数据库，不访问共享环境。 */
+    @BeforeEach
+    void setup() throws Exception {
+        dataSource = new JdbcDataSource();
+        dataSource.setURL("jdbc:h2:mem:completion-lease-" + UUID.randomUUID()
+            + ";MODE=MySQL;DB_CLOSE_DELAY=-1;LOCK_TIMEOUT=1000");
+        jdbc = new JdbcTemplate(dataSource);
+        jdbc.execute("""
+            CREATE TABLE fl_loading_completion_command(
+              id BIGINT PRIMARY KEY,command_status VARCHAR(16),lease_owner VARCHAR(128),lease_until TIMESTAMP,
+              attempt_count INT,next_retry_at TIMESTAMP,updated_at TIMESTAMP,completed_at TIMESTAMP,
+              failure_stage VARCHAR(40),failure_code VARCHAR(32),failure_request_id VARCHAR(64),failure_trace_id VARCHAR(64))
+            """);
+        jdbc.execute("""
+            CREATE TABLE fl_loading_completion_checkpoint(
+              command_id BIGINT,tenant_id BIGINT,inbound_id BIGINT,owner_sequence INT,checkpoint_stage VARCHAR(24),
+              updated_at TIMESTAMP,PRIMARY KEY(command_id,tenant_id,inbound_id))
+            """);
+        jdbc.update("""
+            INSERT INTO fl_loading_completion_command(id,command_status,lease_owner,lease_until,attempt_count,next_retry_at)
+            VALUES(900,'PROCESSING','old-owner',DATEADD('HOUR',1,CURRENT_TIMESTAMP),1,CURRENT_TIMESTAMP)
+            """);
+        var configuration = new MybatisConfiguration();
+        configuration.setMapUnderscoreToCamelCase(true);
+        configuration.addMapper(LoadingCompletionCommandMapper.class);
+        var factory = new MybatisSqlSessionFactoryBean();
+        factory.setDataSource(dataSource);
+        factory.setConfiguration(configuration);
+        mapper = new SqlSessionTemplate(factory.getObject()).getMapper(LoadingCompletionCommandMapper.class);
+        commands = repository(mapper);
+    }
+
+    /** 生产 Repository 的注解事务必须真实生效，覆盖锁定与写入两条 SQL。 */
+    private LoadingCompletionCommandRepository repository(LoadingCompletionCommandMapper targetMapper) {
+        var target = new LoadingCompletionCommandRepository();
+        ReflectionTestUtils.setField(target, "mapper", targetMapper);
+        var proxy = new ProxyFactory(target);
+        proxy.addAdvice(new TransactionInterceptor(new DataSourceTransactionManager(dataSource),
+            new AnnotationTransactionAttributeSource()));
+        return (LoadingCompletionCommandRepository) proxy.getProxy();
+    }
+
+    /** 模拟原执行者续租后暂停，租约过期并被新执行者领取推进，再恢复旧线程。 */
+    @Test
+    void pausedOldOwnerCannotDowngradeTheNewOwnersCompletedCheckpoint() throws Exception {
+        var renewed = new CountDownLatch(1);
+        var resumeOldOwner = new CountDownLatch(1);
+        var executor = Executors.newSingleThreadExecutor();
+        try {
+            var oldWrite = executor.submit(() -> {
+                assertThat(commands.renew(900, "old-owner", LocalDateTime.now().plusMinutes(10))).isTrue();
+                renewed.countDown();
+                assertThat(resumeOldOwner.await(10, TimeUnit.SECONDS)).isTrue();
+                return commands.checkpoint(900, "old-owner", 101, 50, 1, "READY");
+            });
+            assertThat(renewed.await(10, TimeUnit.SECONDS)).isTrue();
+            jdbc.update("UPDATE fl_loading_completion_command SET lease_until=DATEADD('SECOND',-1,CURRENT_TIMESTAMP) WHERE id=900");
+            assertThat(commands.claim(900, "new-owner", LocalDateTime.now().plusMinutes(10))).isTrue();
+            assertThat(commands.checkpoint(900, "new-owner", 101, 50, 1, "INBOUND_COMPLETED")).isTrue();
+            resumeOldOwner.countDown();
+            assertThat(oldWrite.get(10, TimeUnit.SECONDS)).isFalse();
+            assertThat(mapper.checkpointCompleted(900, "old-owner", 101, 50, 1, "SETTLED")).isZero();
+            assertThat(commands.checkpoint(900, 101, 50)).isEqualTo("INBOUND_COMPLETED");
+            assertThat(jdbc.queryForObject("SELECT lease_owner FROM fl_loading_completion_command WHERE id=900", String.class))
+                .isEqualTo("new-owner");
+        } finally {
+            resumeOldOwner.countDown();
+            executor.shutdownNow();
+        }
+    }
+
+    /** 锁定后的实际写入前，独立数据库连接仍不能取得同一命令行的锁。 */
+    @Test
+    void commandRowRemainsLockedAcrossTheLeaseCheckAndCheckpointInsert() {
+        LoadingCompletionCommandMapper observed = mock(LoadingCompletionCommandMapper.class, delegatesTo(mapper));
+        doAnswer(call -> {
+            assertThatThrownBy(() -> {
+                try (var connection = dataSource.getConnection();
+                     var statement = connection.prepareStatement("SELECT id FROM fl_loading_completion_command WHERE id=900 FOR UPDATE NOWAIT")) {
+                    statement.executeQuery().close();
+                }
+            }).isInstanceOf(SQLException.class)
+                .satisfies(error -> assertThat(((SQLException) error).getErrorCode()).isEqualTo(50200));
+            return mapper.checkpointCompleted(call.getArgument(0), call.getArgument(1), call.getArgument(2),
+                call.getArgument(3), call.getArgument(4), call.getArgument(5));
+        }).when(observed).checkpointCompleted(anyLong(), anyString(), anyLong(), anyLong(), anyInt(), anyString());
+        assertThat(repository(observed).checkpoint(900, "old-owner", 101, 50, 1, "READY")).isTrue();
+        assertThat(commands.checkpoint(900, 101, 50)).isEqualTo("READY");
+    }
+
+    /** 初次锁定检查后若租约失效，实际 INSERT 也必须拒绝，不能仅依赖前一次检查。 */
+    @Test
+    void checkpointInsertRechecksLeaseValidityAfterTheInitialLock() {
+        LoadingCompletionCommandMapper observed = mock(LoadingCompletionCommandMapper.class, delegatesTo(mapper));
+        doAnswer(call -> {
+            jdbc.update("UPDATE fl_loading_completion_command SET lease_until=DATEADD('SECOND',-1,CURRENT_TIMESTAMP) WHERE id=900");
+            return mapper.checkpointCompleted(call.getArgument(0), call.getArgument(1), call.getArgument(2),
+                call.getArgument(3), call.getArgument(4), call.getArgument(5));
+        }).when(observed).checkpointCompleted(anyLong(), anyString(), anyLong(), anyLong(), anyInt(), anyString());
+        assertThat(repository(observed).checkpoint(900, "old-owner", 101, 50, 1, "READY")).isFalse();
+        assertThat(commands.checkpoint(900, 101, 50)).isNull();
+    }
+
+    /** 驱动将已持久化的幂等重放计为零行时，仍须以锁内当前有效租约确认。 */
+    @Test
+    void zeroAffectedRowsCanConfirmAnIdempotentCheckpointUnderTheSameValidLease() {
+        assertThat(commands.checkpoint(900, "old-owner", 101, 50, 1, "READY")).isTrue();
+        LoadingCompletionCommandMapper observed = mock(LoadingCompletionCommandMapper.class, delegatesTo(mapper));
+        doAnswer(call -> {
+            mapper.checkpointCompleted(call.getArgument(0), call.getArgument(1), call.getArgument(2),
+                call.getArgument(3), call.getArgument(4), call.getArgument(5));
+            return 0;
+        }).when(observed).checkpointCompleted(anyLong(), anyString(), anyLong(), anyLong(), anyInt(), anyString());
+        assertThat(repository(observed).checkpoint(900, "old-owner", 101, 50, 1, "READY")).isTrue();
+        assertThat(commands.checkpoint(900, 101, 50)).isEqualTo("READY");
+    }
+
+    /** 已过期但尚未被重新领取的所有者也不能写检查点、续租或写命令终态。 */
+    @Test
+    void expiredLeaseCannotWriteCheckpointRenewFailOrComplete() {
+        jdbc.update("UPDATE fl_loading_completion_command SET lease_until=DATEADD('SECOND',-1,CURRENT_TIMESTAMP) WHERE id=900");
+        assertThat(commands.checkpoint(900, "old-owner", 101, 50, 1, "READY")).isFalse();
+        assertThat(mapper.checkpointCompleted(900, "old-owner", 101, 50, 1, "READY")).isZero();
+        assertThat(commands.renew(900, "old-owner", LocalDateTime.now().plusMinutes(10))).isFalse();
+        assertThat(commands.fail(900, "old-owner", "FAILED", LocalDateTime.now(), "LEASE_LOST", "A0409", null, null)).isFalse();
+        assertThat(commands.complete(900, "old-owner")).isFalse();
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM fl_loading_completion_checkpoint", Integer.class)).isZero();
+        assertThat(jdbc.queryForObject("SELECT command_status FROM fl_loading_completion_command WHERE id=900", String.class))
+            .isEqualTo("PROCESSING");
+    }
+
+    /** 同一当前所有者的重复或迟到阶段也只能保持或推进，不能降低完成检查点。 */
+    @Test
+    void validOwnerCheckpointStagesAreMonotonic() {
+        assertThat(commands.checkpoint(900, "old-owner", 101, 50, 1, "SETTLED")).isTrue();
+        commands.checkpoint(900, "old-owner", 101, 50, 1, "READY");
+        assertThat(commands.checkpoint(900, 101, 50)).isEqualTo("SETTLED");
+        assertThat(commands.checkpoint(900, "old-owner", 101, 50, 1, "INBOUND_COMPLETED")).isTrue();
+        commands.checkpoint(900, "old-owner", 101, 50, 1, "SETTLED");
+        commands.checkpoint(900, "old-owner", 101, 50, 1, "READY");
+        assertThat(commands.checkpoint(900, 101, 50)).isEqualTo("INBOUND_COMPLETED");
+    }
+
+    /** 完成或失败后的命令不允许追加检查点，即使所有者与租期字段仍有旧值。 */
+    @Test
+    void terminalCommandCannotAcceptAnotherCheckpoint() {
+        jdbc.update("UPDATE fl_loading_completion_command SET command_status='COMPLETED' WHERE id=900");
+        assertThat(commands.checkpoint(900, "old-owner", 101, 50, 1, "READY")).isFalse();
+        assertThat(mapper.checkpointCompleted(900, "old-owner", 101, 50, 1, "READY")).isZero();
+    }
+}
@@ -31,6 +31,8 @@ class LoadingCompletionOwnerPersistenceTest {
        jdbc.execute("CREATE TABLE fl_loading_plan_item(id BIGINT,loading_plan_id BIGINT,account_id BIGINT,inbound_id BIGINT,deleted INT,package_count INT)");
        jdbc.execute("CREATE TABLE fl_loading_pick_item(id BIGINT,pick_task_id BIGINT,loading_plan_item_id BIGINT,item_status VARCHAR,picked_package_count INT)");
        jdbc.execute("CREATE TABLE fl_loading_completion_checkpoint(command_id BIGINT,tenant_id BIGINT,inbound_id BIGINT,owner_sequence INT,checkpoint_stage VARCHAR(24),updated_at TIMESTAMP,PRIMARY KEY(command_id,tenant_id,inbound_id))");
+        jdbc.execute("CREATE TABLE fl_loading_completion_command(id BIGINT PRIMARY KEY,command_status VARCHAR(16),lease_owner VARCHAR(128),lease_until TIMESTAMP)");
+        jdbc.update("INSERT INTO fl_loading_completion_command VALUES(900,'PROCESSING','owner-test',DATEADD('HOUR',1,CURRENT_TIMESTAMP))");
        jdbc.update("INSERT INTO fl_loading_pick_task VALUES(20,10)");
        row(1, 101, 50, 3, "PARTIAL", 2, 10, 0);
        row(2, 101, 50, 1, "SKIPPED", 0, 10, 0); // same source SKU history
@@ -76,11 +78,11 @@ class LoadingCompletionOwnerPersistenceTest {

    @Test
    void completedActualCheckpointsAreReadBackUnchangedAndSkippedReadyCheckpointIsNotACompletionOwner() {
-        checkpoints.checkpointCompleted(900, 101, 50, 1, "READY");
-        checkpoints.checkpointCompleted(900, 101, 50, 1, "SETTLED");
-        checkpoints.checkpointCompleted(900, 101, 50, 1, "INBOUND_COMPLETED");
-        checkpoints.checkpointCompleted(900, 202, 60, 2, "INBOUND_COMPLETED");
-        checkpoints.checkpointCompleted(900, 303, 70, 3, "READY");
+        checkpoints.checkpointCompleted(900, "owner-test", 101, 50, 1, "READY");
+        checkpoints.checkpointCompleted(900, "owner-test", 101, 50, 1, "SETTLED");
+        checkpoints.checkpointCompleted(900, "owner-test", 101, 50, 1, "INBOUND_COMPLETED");
+        checkpoints.checkpointCompleted(900, "owner-test", 202, 60, 2, "INBOUND_COMPLETED");
+        checkpoints.checkpointCompleted(900, "owner-test", 303, 70, 3, "READY");
        assertThat(checkpoints.checkpoint(900, 101, 50)).isEqualTo("INBOUND_COMPLETED");
        assertThat(checkpoints.checkpoint(900, 202, 60)).isEqualTo("INBOUND_COMPLETED");
        assertThat(checkpoints.checkpoint(900, 303, 70)).isEqualTo("READY");
@@ -8,22 +8,124 @@ 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 org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.task.TaskExecutor;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.annotation.EnableTransactionManagement;
+import org.springframework.transaction.support.AbstractPlatformTransactionManager;
+import org.springframework.transaction.support.DefaultTransactionStatus;
+import org.springframework.transaction.support.TransactionTemplate;
+import org.mockito.ArgumentCaptor;

 import java.time.LocalDateTime;
 import java.util.List;
+import java.util.concurrent.RejectedExecutionException;

 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;

-@SpringJUnitConfig(LoadingCompletionWorker.class)
+@SpringJUnitConfig({LoadingCompletionWorker.class, LoadingCompletionWorkerTest.TransactionEvents.class})
 class LoadingCompletionWorkerTest {
+    /** 与生产事务装配一致，注册事务事件工厂，避免普通事件工厂立即调用监听器。 */
+    @Configuration
+    @EnableTransactionManagement
+    static class TransactionEvents {}
+
    @MockBean private LoadingCompletionCommandRepository commands;
    @MockBean private LoadingPickRepository picks;
    @MockBean private LoadingPickService loadingPicks;
+    @MockBean(name = "applicationTaskExecutor") private TaskExecutor completionExecutor;
    @jakarta.annotation.Resource private LoadingCompletionWorker worker;
+    @jakarta.annotation.Resource private ApplicationEventPublisher events;
+
+    /** 真实事务同步回调必须在提交后才将完成工作交给执行器。 */
+    @Test
+    void committedAcceptanceStartsImmediatelyThroughTheDurableLease() {
+        LoadingCompletionCommandEntity command = command(0);
+        when(commands.command(900L)).thenReturn(command);
+        when(commands.claim(eq(900L), any(String.class), any(LocalDateTime.class))).thenReturn(true);
+        transactions().executeWithoutResult(status -> {
+            events.publishEvent(new LoadingPickService.CompletionAccepted(900L, System.nanoTime()));
+            verify(completionExecutor, never()).execute(any());
+            verify(loadingPicks, never()).processCompletion(any(), any());
+        });
+
+        ArgumentCaptor<Runnable> work = ArgumentCaptor.forClass(Runnable.class);
+        verify(completionExecutor).execute(work.capture());
+        verify(loadingPicks, never()).processCompletion(any(), any());
+        work.getValue().run();
+
+        verify(commands).claim(eq(900L), any(String.class), any(LocalDateTime.class));
+        verify(loadingPicks).processCompletion(eq(command), any(String.class));
+    }
+
+    /** 回滚后的命令没有权威落库事实，不允许唤醒或执行结算。 */
+    @Test
+    void rolledBackAcceptanceNeverSchedulesCompletion() {
+        transactions().executeWithoutResult(status -> {
+            events.publishEvent(new LoadingPickService.CompletionAccepted(900L, System.nanoTime()));
+            status.setRollbackOnly();
+        });
+        verify(completionExecutor, never()).execute(any());
+        verify(commands, never()).command(900L);
+    }
+
+    /** 即时唤醒不能绕过定时执行者已取得的租约。 */
+    @Test
+    void immediateWakeupDoesNotProcessAnAlreadyLeasedCommand() {
+        when(commands.command(900L)).thenReturn(command(0));
+        worker.accepted(new LoadingPickService.CompletionAccepted(900L, System.nanoTime()));
+        ArgumentCaptor<Runnable> work = ArgumentCaptor.forClass(Runnable.class);
+        verify(completionExecutor).execute(work.capture());
+        work.getValue().run();
+        verify(loadingPicks, never()).processCompletion(any(), any());
+    }
+
+    /** 执行器拒绝时受理不报错，定时扫描仍可从原命令恢复。 */
+    @Test
+    void rejectedWakeupLeavesScheduledRecoveryAvailable() {
+        doThrow(new RejectedExecutionException("busy")).when(completionExecutor).execute(any());
+        LoadingCompletionCommandEntity command = command(0);
+        when(commands.due(10)).thenReturn(List.of(command));
+        when(commands.claim(eq(900L), any(String.class), any(LocalDateTime.class))).thenReturn(true);
+        worker.accepted(new LoadingPickService.CompletionAccepted(900L, System.nanoTime()));
+        worker.dispatch();
+        verify(loadingPicks).processCompletion(eq(command), any(String.class));
+    }
+
+    /** 同进程重领已到期命令也必须更换所有者，旧线程不能续租或落库。 */
+    @Test
+    void everyClaimUsesAnIndependentLeaseOwner() {
+        LoadingCompletionCommandEntity command = command(0);
+        when(commands.due(10)).thenReturn(List.of(command));
+        when(commands.claim(eq(900L), any(String.class), any(LocalDateTime.class))).thenReturn(true);
+        worker.dispatch();
+        worker.dispatch();
+        ArgumentCaptor<String> owners = ArgumentCaptor.forClass(String.class);
+        verify(loadingPicks, times(2)).processCompletion(eq(command), owners.capture());
+        assertThat(owners.getAllValues()).doesNotHaveDuplicates();
+    }
+
+    /** 隔离数据库依赖，仅用 Spring 的真实事务同步机制核对提交、回滚边界。 */
+    private static TransactionTemplate transactions() {
+        return new TransactionTemplate(new AbstractPlatformTransactionManager() {
+            /** 为每次测试提供独立事务对象。 */
+            @Override protected Object doGetTransaction() { return new Object(); }
+            /** 测试事务不访问数据库。 */
+            @Override protected void doBegin(Object transaction, TransactionDefinition definition) {}
+            /** 让 Spring 正常触发提交同步回调。 */
+            @Override protected void doCommit(DefaultTransactionStatus status) {}
+            /** 让 Spring 正常触发回滚同步回调。 */
+            @Override protected void doRollback(DefaultTransactionStatus status) {}
+        });
+    }

    @Test
    void transientFailureIsReleasedForBackoffRetry() {
@@ -50,6 +50,9 @@ import org.mockito.InOrder;
 import org.mockito.ArgumentCaptor;
 import org.springframework.boot.test.mock.mockito.MockBean;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+import org.springframework.test.context.event.RecordApplicationEvents;
+import org.springframework.test.context.event.ApplicationEvents;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.transaction.TransactionStatus;
 import org.springframework.transaction.support.TransactionTemplate;

@@ -77,7 +80,9 @@ import static org.mockito.Mockito.clearInvocations;
 import static org.assertj.core.api.Assertions.assertThat;

@SpringJUnitConfig(LoadingPickService.class)
+@RecordApplicationEvents
 class LoadingPickServiceTest {
+    @Autowired private ApplicationEvents completionEvents;
    @MockBean
    private LoadingPickRepository repository;
    @MockBean
@@ -154,6 +159,8 @@ class LoadingPickServiceTest {
        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(completionCommands.checkpoint(anyLong(), anyString(), anyLong(), anyLong(), anyInt(), anyString()))
+            .thenReturn(true);
        when(repository.acceptCompletion(anyLong(), anyInt())).thenReturn(1);
        when(repository.inboundOwners(anyLong())).thenReturn(List.of(
            new LoadingPickItemMapper.InboundOwner(101L, 50L)));
@@ -931,6 +938,7 @@ class LoadingPickServiceTest {
        assertThat(replay.taskStatus()).isEqualTo("COMPLETING");
        verify(repository, never()).taskForUpdate(20L);
        verify(completionCommands, times(1)).insert(any());
+        assertThat(completionEvents.stream(LoadingPickService.CompletionAccepted.class).count()).isEqualTo(1);
        verify(settlementService, never()).settleOwner(anyLong(), anyLong(), anyLong(), anyString());
        verify(loadingListService, never()).finishFromPda(anyLong(), anyLong(), anyLong(), anyString());
    }
@@ -997,6 +1005,8 @@ class LoadingPickServiceTest {

        assertThat(result.workflowStatus()).isEqualTo("PROCESSING");
        verify(completionCommands).resume(902L);
+        assertThat(completionEvents.stream(LoadingPickService.CompletionAccepted.class)
+            .map(LoadingPickService.CompletionAccepted::commandId)).containsExactly(902L);
        verify(repository).resumeCompletion(20L);
        verify(completionCommands, never()).insert(any());
    }
@@ -1089,7 +1099,7 @@ class LoadingPickServiceTest {
        verify(settlementService, never()).settleOwner(anyLong(), anyLong(), anyLong(), anyString());
        verify(loadingDeliveryService, never()).completeAfterPdaSettlement(anyLong(), anyLong(), anyLong(), anyString(), anyString(), anyString());
        verify(completionCommands, never()).checkpoint(eq(931L), eq(303L), eq(70L));
-        verify(completionCommands, never()).checkpoint(eq(931L), anyLong(), anyLong(), anyInt(), anyString());
+        verify(completionCommands, never()).checkpoint(eq(931L), anyString(), anyLong(), anyLong(), anyInt(), anyString());
        verify(loadingListService).publishCompletedEvent(10L);
        verify(completionCommands).complete(931L, "worker-completed-checkpoints");
    }
@@ -1109,6 +1119,31 @@ class LoadingPickServiceTest {
        verify(loadingListService, never()).publishCompletedEvent(anyLong());
    }

+    /** 任一检查点被租约拒绝后都不得继续下一业务步骤或最终完成柜次。 */
+    @ParameterizedTest
+    @ValueSource(strings = {"READY", "SETTLED", "INBOUND_COMPLETED"})
+    void lostLeaseAtCheckpointStopsFurtherCompletion(String rejectedStage) {
+        LoadingPickTaskEntity completing = task();
+        completing.setTaskStatus("COMPLETING");
+        when(repository.task(20L)).thenReturn(completing);
+        when(completionCommands.checkpoint(eq(940L), eq("old-owner"), eq(101L), eq(50L), eq(1), eq(rejectedStage)))
+            .thenReturn(false);
+
+        assertThatThrownBy(() -> service.processCompletion(completionCommand(940L, "lost-checkpoint"), "old-owner"))
+            .isInstanceOf(LoadingPickService.CompletionProcessingException.class)
+            .satisfies(error -> assertThat(((LoadingPickService.CompletionProcessingException) error).stage())
+                .isEqualTo("LEASE_LOST"));
+        if ("READY".equals(rejectedStage)) {
+            verify(settlementService, never()).settleOwner(anyLong(), anyLong(), anyLong(), anyString());
+        }
+        if (!"INBOUND_COMPLETED".equals(rejectedStage)) {
+            verify(loadingDeliveryService, never()).completeAfterPdaSettlement(anyLong(), anyLong(), anyLong(),
+                anyString(), anyString(), anyString());
+        }
+        verify(loadingListService, never()).finishFromPda(anyLong(), anyLong(), anyLong(), anyString());
+        verify(completionCommands, never()).complete(anyLong(), anyString());
+    }
+
    @Test
    void anEntirelySkippedTaskCannotCreateACompletionCommand() {
        when(repository.taskForUpdate(20L)).thenReturn(task());
@@ -1271,8 +1306,8 @@ class LoadingPickServiceTest {
        verify(settlementService, never()).settleOwner(anyLong(), anyLong(), anyLong(), anyString());
        verify(loadingDeliveryService, never()).completeAfterPdaSettlement(anyLong(), anyLong(), anyLong(),
            anyString(), anyString(), anyString());
-        verify(completionCommands).checkpoint(904L, 101L, 50L, 1, "READY");
-        verify(completionCommands, never()).checkpoint(eq(904L), anyLong(), anyLong(), anyInt(),
+        verify(completionCommands).checkpoint(904L, "worker-failure", 101L, 50L, 1, "READY");
+        verify(completionCommands, never()).checkpoint(eq(904L), anyString(), anyLong(), anyLong(), anyInt(),
            eq("SETTLED"));
        verify(repository, never()).insert(any(LoadingPickOperationEntity.class));
        verify(loadingListService, never()).finishFromPda(anyLong(), anyLong(), anyLong(), anyString());
@@ -1300,7 +1335,7 @@ class LoadingPickServiceTest {
        verify(settlementService, never()).settleOwner(anyLong(), anyLong(), anyLong(), anyString());
        verify(loadingDeliveryService, never()).completeAfterPdaSettlement(anyLong(), anyLong(), anyLong(),
            anyString(), anyString(), anyString());
-        verify(completionCommands, never()).checkpoint(anyLong(), anyLong(), anyLong(), anyInt(), anyString());
+        verify(completionCommands, never()).checkpoint(anyLong(), anyString(), anyLong(), anyLong(), anyInt(), anyString());
        verify(repository, never()).insert(any(LoadingPickOperationEntity.class));
        verify(loadingListService, never()).finishFromPda(anyLong(), anyLong(), anyLong(), anyString());
        verify(loadingListService, never()).publishCompletedEvent(anyLong());
@@ -46,6 +46,8 @@ class ContainerCalendarSourceMapperTest {
            var calendar = session.getMapper(ContainerCalendarSourceMapper.class).sources(from, from.minusDays(2), from.minusDays(4), from.plusMonths(1));
            var workbench = session.getMapper(CarrierScheduleQueryMapper.class).selectContainerActualFacts(java.util.List.of(1L, 2L));
            assertEquals(2, calendar.size());
+            var explicit = session.getMapper(ContainerCalendarSourceMapper.class).byIds(java.util.List.of(1L,2L));
+            assertEquals(calendar,explicit);
            assertEquals(2, workbench.size());
            for (int index = 0; index < calendar.size(); index++) {
                var expected = LocalDateTime.parse(index == 0 ? "2026-10-21T14:00:00" : "2026-10-22T14:00:00");