perf(lastleg): limit transport reconciliation facts and trace sync phases

2026-10-17 09:11:00 +08:00
parent 732cf99007
commit 81e4123b10
2 changed files with 119 additions and 6 deletions
@@ -2,6 +2,8 @@ package com.baiyue.lastleg.appointmentwork;

 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import java.time.*;
 import java.util.*;
 import java.util.function.Function;
@@ -12,6 +14,7 @@ import static com.baiyue.lastleg.appointmentwork.AppointmentWorkPolicy.*;

@Service
 public class AppointmentWorkSyncService {
+    private static final Logger LOG=LoggerFactory.getLogger(AppointmentWorkSyncService.class);
    private static final Actor SYSTEM=new Actor("appointment-work-sync","服务端工作范围同步",Set.of(LIST,ORDER_READ));
    private final AppointmentWorkStore store;
    private final AppointmentWorkFactGateway gateway;
@@ -20,12 +23,17 @@ public class AppointmentWorkSyncService {
    @Transactional
    public SyncResult synchronize(SyncRequest request) {
        if(request==null || request.size()<1 || request.size()>100) throw bad("同步批次须为1至100柜");
+        var timing=new SyncTiming();boolean success=false;
+        try {
        var guard=store.lock();
        if(request.afterContainerBookingId()!=null && !Objects.equals(guard.getScanCursor(),request.afterContainerBookingId())) throw conflict("来源同步游标已变化，请从首批重新核对");
+        timing.next(SyncPhase.DISCOVER);
        var page=gateway.discover(request.afterContainerBookingId(),request.size());
+        timing.pageContainers=page.containers().size();timing.complete=page.complete();
        if(request.afterContainerBookingId()==null) { guard.setScanScopeVersion(page.scopeVersion()); guard.setScanRequirementVersion(guard.getRequirementVersion()); }
        else if(!Objects.equals(guard.getScanScopeVersion(),page.scopeVersion())) throw conflict("来源范围扫描中变化，请从首批重试");
        else if(guard.getScanRequirementVersion()!=guard.getRequirementVersion()) throw conflict("预约需求扫描中变化，请从首批重试");
+        timing.next(SyncPhase.GENERATE);
        // 保留全部历史与稳定键顺序；逐柜更新立即进入本次工作集，不跨事务缓存。
        var known=workingSet();
        int created=0; var ids=new HashSet<String>();
@@ -51,18 +59,54 @@ public class AppointmentWorkSyncService {
                }
            }
        }
-        // An existing row is never deleted because an order left the active source page.
-        // Empty return and other unfinished work continue to read their original authority.
+        // 原来源离开当前页也保留历史事项，未完成工作仍在第二轮按原业务完整核对。
+        timing.next(SyncPhase.TRANSPORT_FACTS);
        var all=new ArrayList<>(store.all());
-        var facts=gateway.facts(all.stream().filter(s->s.getType()!=Type.ROUTE && s.getTripKey()==null && !"BATCH_REPLACED".equals(s.getRetirementReason())).toList());
+        // 首轮仅归并运输关系；保留已关联车次的退役成员，才能发现解绑并恢复原运输事项。
+        var transportItems=all.stream().filter(s->s.getType()==Type.TRANSPORT && s.getTripKey()==null && !"BATCH_REPLACED".equals(s.getRetirementReason())).toList();
+        timing.transportFactsCount=transportItems.size();
+        var facts=gateway.facts(transportItems);
+        timing.next(SyncPhase.RECONCILE);
        reconcileTransport(all,facts);
+        timing.next(SyncPhase.ACTIVE_FACTS);
        all=new ArrayList<>(store.all());
-        facts=gateway.facts(all.stream().filter(s->!s.isRetired()).toList());
+        var activeItems=all.stream().filter(s->!s.isRetired()).toList();
+        timing.activeFactsCount=activeItems.size();
+        facts=gateway.facts(activeItems);
+        timing.next(SyncPhase.OBSERVE_CHECKPOINT);
        observe(all,facts,SYSTEM);
        guard.setScanCursor(page.complete()?null:page.nextCursor());
        if(page.complete()) { guard.setInitialized(true); guard.setScopeVersion(page.scopeVersion()); guard.setSynchronizedRequirementVersion(guard.getRequirementVersion()); guard.setCompletedAt(LocalDateTime.now(ZoneOffset.UTC)); }
        store.checkpoint(guard);
+        success=true;
        return new SyncResult(page.containers().size(),created,page.nextCursor(),page.complete());
+        } finally { timing.finish(success); }
+    }
+    /** 固定阶段名用于慢同步定位，不包含来源、游标或异常业务内容。 */
+    private enum SyncPhase { GUARD, DISCOVER, GENERATE, TRANSPORT_FACTS, RECONCILE, ACTIVE_FACTS, OBSERVE_CHECKPOINT }
+    /** 仅统计本次方法执行，不含事务代理提交，不能当作精确锁持有时间。 */
+    private static final class SyncTiming {
+        private final long started=System.nanoTime();
+        private final long[] elapsed=new long[SyncPhase.values().length];
+        private long phaseStarted=started;
+        private SyncPhase phase=SyncPhase.GUARD;
+        private int pageContainers=-1,transportFactsCount=-1,activeFactsCount=-1;
+        private boolean complete;
+        /** 完成前一阶段后开始下一阶段；事实阶段包含紧邻的数据库回读。 */
+        private void next(SyncPhase next) {
+            long now=System.nanoTime();elapsed[phase.ordinal()]+=now-phaseStarted;phaseStarted=now;phase=next;
+        }
+        /** 成功和失败共用一条慢调用摘要，未到达的计数保持为负一。 */
+        private void finish(boolean success) {
+            long now=System.nanoTime();elapsed[phase.ordinal()]+=now-phaseStarted;
+            long totalMs=java.util.concurrent.TimeUnit.NANOSECONDS.toMillis(now-started);
+            if(totalMs<250) return;
+            LOG.warn("Appointment work sync slow: totalMs={}, guardMs={}, discoverMs={}, generateMs={}, transportFactsMs={}, reconcileMs={}, activeFactsMs={}, observeCheckpointMs={}, pageContainers={}, transportFactsCount={}, activeFactsCount={}, complete={}, failedStage={}",
+                totalMs,millis(SyncPhase.GUARD),millis(SyncPhase.DISCOVER),millis(SyncPhase.GENERATE),millis(SyncPhase.TRANSPORT_FACTS),millis(SyncPhase.RECONCILE),millis(SyncPhase.ACTIVE_FACTS),millis(SyncPhase.OBSERVE_CHECKPOINT),
+                pageContainers,transportFactsCount,activeFactsCount,complete,success?"NONE":phase.name());
+        }
+        /** 阶段仅输出整数毫秒，不携带业务标识。 */
+        private long millis(SyncPhase value) { return java.util.concurrent.TimeUnit.NANOSECONDS.toMillis(elapsed[value.ordinal()]); }
    }
    static void validateSource(ContainerSource source) {
        if(source==null || source.containerBookingId()==null || source.containerBookingId().isBlank() || source.departedAt()==null && source.departedLocalAt()==null && source.sailedEvidenceLocalAt()==null || source.sourceVersion()==null || source.establishedFlow()==null || source.requirements()==null)
@@ -207,16 +207,24 @@ class AppointmentWorkSyncServiceTest {
        assertThrows(ResponseStatusException.class,()->f.sync.synchronize(new SyncRequest("10",10)));
        assertFalse(f.guard.isInitialized());
    }
+    /** 首轮只取运输事实；关联与解绑均保留原成员，第二轮仍观察全部活跃工作。 */
    @Test void sharedTripAcrossTwoContainersIsPersistedOnceAndKeepsBothContainers() {
        var f=new WorkTestFixture();
+        var replaced=f.state("old",Type.TRANSPORT,Flow.TRANSFER,"old-requirement");
+        replaced.setRetired(true);replaced.setRetirementReason("BATCH_REPLACED");f.add(replaced);
+        record ScopeItem(String key,Type type,String trip,boolean retired,String reason) { }
+        var scopes=new ArrayList<List<ScopeItem>>();
+        var linked=new java.util.concurrent.atomic.AtomicBoolean(true);
        var sources=List.of("1","2").stream().map(id->new ContainerSource(id,"柜"+id,Instant.parse("2026-10-01T00:00:00Z"),null,null,"v1",Flow.TRANSFER,"America/Los_Angeles",
            List.of(new Requirement("R"+id,"B"+id,"AP"+id,"FBA","海外仓","ONT8","America/Los_Angeles",List.of(),List.of(),List.of())))).toList();
        when(f.gateway.discover(null,10)).thenReturn(new SourcePage(sources,null,true,"scope"));
        when(f.gateway.facts(anyList())).thenAnswer(call->{
+            var requested=call.<List<AppointmentWorkState>>getArgument(0);
+            scopes.add(requested.stream().map(s->new ScopeItem(s.getWorkItemKey(),s.getType(),s.getTripKey(),s.isRetired(),s.getRetirementReason())).toList());
            var values=new HashMap<String,Fact>();
-            for(var state:call.<List<AppointmentWorkState>>getArgument(0)) if(state.getType()!=Type.ROUTE) {
+            for(var state:requested) if(state.getType()!=Type.ROUTE) {
                if(state.getType()==Type.TRANSPORT) values.put(state.getWorkItemKey(),new Fact(state.getWorkItemKey(),"trip-v1","trip-impact-v1",true,PlanStatus.CONFIRMED,ExecutionStatus.PENDING,
-                    slot("2026-10-16"),slot("2026-10-16"),null,"America/Los_Angeles","TR-1","trip-1",List.of(AppointmentWorkPolicy.ORDER_READ),List.of(),null,List.of(),List.of(),List.of(),null,null));
+                    slot("2026-10-16"),slot("2026-10-16"),null,"America/Los_Angeles",linked.get()?"TR-1":null,linked.get()?"trip-1":null,List.of(AppointmentWorkPolicy.ORDER_READ),List.of(),null,List.of(),List.of(),List.of(),null,null));
                else values.put(state.getWorkItemKey(),fact(state,null,null,ExecutionStatus.PENDING,null));
            }
            return values;
@@ -225,5 +233,66 @@ class AppointmentWorkSyncServiceTest {
        var transports=f.store.all().stream().filter(s->!s.isRetired() && s.getType()==Type.TRANSPORT).toList();
        assertEquals(1,transports.size()); assertEquals(2,transports.get(0).getContainers().size()); assertEquals(List.of("B1","B2"),transports.get(0).getBatchKeys());
        assertEquals(9,f.store.all().stream().filter(s->!s.isRetired()).count());
+        assertEquals(4,scopes.size());
+        for(int index:List.of(0,2)) {
+            assertEquals(2,scopes.get(index).size());
+            assertTrue(scopes.get(index).stream().allMatch(s->s.type()==Type.TRANSPORT && s.trip()==null && !"BATCH_REPLACED".equals(s.reason())));
+        }
+        assertTrue(scopes.get(2).stream().allMatch(s->s.retired() && "TRIP_LINKED".equals(s.reason())));
+        var activeKeys=f.store.all().stream().filter(s->!s.isRetired()).map(AppointmentWorkState::getWorkItemKey).collect(java.util.stream.Collectors.toSet());
+        for(int index:List.of(1,3)) assertEquals(activeKeys,scopes.get(index).stream().map(ScopeItem::key).collect(java.util.stream.Collectors.toSet()));
+        linked.set(false);scopes.clear();
+        f.sync.synchronize(new SyncRequest(null,10));
+        assertEquals(2,scopes.size());assertEquals(2,scopes.get(0).size());
+        assertTrue(scopes.get(0).stream().allMatch(s->s.type()==Type.TRANSPORT && s.retired() && "TRIP_LINKED".equals(s.reason())));
+        var restored=f.store.all().stream().filter(s->!s.isRetired() && s.getType()==Type.TRANSPORT).toList();
+        assertEquals(2,restored.size());assertTrue(restored.stream().allMatch(s->s.getTripKey()==null && s.getReplacementKeys().isEmpty()));
+        assertEquals("TRIP_UNLINKED",f.store.get(transports.get(0).getWorkItemKey()).getRetirementReason());
+        assertEquals(f.store.all().stream().filter(s->!s.isRetired()).map(AppointmentWorkState::getWorkItemKey).collect(java.util.stream.Collectors.toSet()),
+            scopes.get(1).stream().map(ScopeItem::key).collect(java.util.stream.Collectors.toSet()));
+        assertTrue(f.store.get(replaced.getWorkItemKey()).isRetired());
+    }
+
+    /** 任一事实阶段失败均不得发布完整范围；此模拟验证发布门禁，不声称数据库回滚。 */
+    @Test void failureInEitherFactRoundDoesNotPublishSynchronizedScope() {
+        for(int failedRound:List.of(1,2)) {
+            var f=new WorkTestFixture();f.guard.setInitialized(false);f.guard.setRequirementVersion(7);f.guard.setSynchronizedRequirementVersion(6);
+            var transport=f.state("1",Type.TRANSPORT,Flow.TRANSFER,"r");f.add(transport);
+            var pickup=f.state("1",Type.PICKUP,Flow.TRANSFER,null);f.add(pickup);
+            var calls=new java.util.concurrent.atomic.AtomicInteger();
+            when(f.gateway.facts(anyList())).thenAnswer(call->{
+                var requested=call.<List<AppointmentWorkState>>getArgument(0);int round=calls.incrementAndGet();
+                if(round==1) assertEquals(List.of(transport.getWorkItemKey()),requested.stream().map(AppointmentWorkState::getWorkItemKey).toList());
+                else assertTrue(requested.stream().anyMatch(s->s.getWorkItemKey().equals(pickup.getWorkItemKey())));
+                if(round==failedRound) throw AppointmentWorkPolicy.unavailable("原业务事实暂不可用");
+                return Map.of(transport.getWorkItemKey(),fact(transport,null,null,ExecutionStatus.PENDING,null));
+            });
+            assertThrows(ResponseStatusException.class,()->f.sync.synchronize(new SyncRequest(null,10)));
+            assertEquals(failedRound,calls.get());assertFalse(f.guard.isInitialized());
+            assertEquals(6,f.guard.getSynchronizedRequirementVersion());assertNull(f.guard.getCompletedAt());
+            verify(f.mapper,never()).saveGuard(any());
+        }
+    }
+
+    /** 慢事实失败保留原异常，只记录一次固定阶段摘要，不输出异常业务内容。 */
+    @Test void slowFactFailureLogsOnePhaseSummaryWithoutPublishingOrLeakingFailure() {
+        var f=new WorkTestFixture();var calls=new java.util.concurrent.atomic.AtomicInteger();
+        var failure=AppointmentWorkPolicy.unavailable("private-source-detail-must-not-be-logged");
+        when(f.gateway.facts(anyList())).thenAnswer(call->{
+            if(calls.incrementAndGet()==1) return Map.of();
+            Thread.sleep(270);throw failure;
+        });
+        var logger=(ch.qos.logback.classic.Logger)org.slf4j.LoggerFactory.getLogger(AppointmentWorkSyncService.class);
+        var appender=new ch.qos.logback.core.read.ListAppender<ch.qos.logback.classic.spi.ILoggingEvent>();
+        appender.start();logger.addAppender(appender);
+        try {
+            assertSame(failure,assertThrows(ResponseStatusException.class,()->f.sync.synchronize(new SyncRequest(null,10))));
+            verify(f.mapper,never()).saveGuard(any());
+            assertEquals(1,appender.list.size());var event=appender.list.get(0);
+            assertEquals(ch.qos.logback.classic.Level.WARN,event.getLevel());
+            assertTrue(event.getFormattedMessage().contains("failedStage=ACTIVE_FACTS"));
+            assertTrue(event.getFormattedMessage().contains("transportFactsCount=0, activeFactsCount=0"));
+            assertFalse(event.getFormattedMessage().contains("private-source-detail"));assertNull(event.getThrowableProxy());
+        } finally { logger.detachAppender(appender);appender.stop(); }
    }
 }