perf(lastleg): reuse work items within synchronization batch

2026-10-17 08:30:11 +08:00
parent 5fd74e9c8c
commit 732cf99007
3 changed files with 84 additions and 9 deletions
@@ -31,8 +31,21 @@ public class AppointmentWorkStore {
            }
        }
    }
+    /** 测量守卫读取加锁耗时，保留原有锁和失败行为，覆盖尚未进入更新语句的慢命令。 */
    public AppointmentWorkMapper.Guard lock() {
-        var guard=mapper.lockGuard(); if(guard==null) throw unavailable("工作事项DDL尚未集成"); return guard;
+        long started = System.nanoTime();
+        try {
+            var guard = mapper.lockGuard();
+            if (guard == null) throw unavailable("工作事项DDL尚未集成");
+            return guard;
+        } finally {
+            long elapsedMs = java.util.concurrent.TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - started);
+            if (elapsedMs >= 250) {
+                // 耗时包含连接及语句执行，单条日志不能区分具体等待原因，不记录业务载荷。
+                LOG.warn("Appointment work guard lock slow: elapsedMs={}, requestId={}, traceId={}",
+                    elapsedMs, MDC.get("requestId"), MDC.get("traceId"));
+            }
+        }
    }
    public String requireInitialized() {
        var guard=mapper.guard();
@@ -16,7 +16,7 @@ public class AppointmentWorkSyncService {
    private final AppointmentWorkStore store;
    private final AppointmentWorkFactGateway gateway;
    public AppointmentWorkSyncService(AppointmentWorkStore store,AppointmentWorkFactGateway gateway) { this.store=store; this.gateway=gateway; }
-    /** Called by trusted server orchestration, never by GET or by opening a particular container. */
+    /** 服务端按完整范围同步；同一守卫事务内复用工作集，GET 不触发写入。 */
    @Transactional
    public SyncResult synchronize(SyncRequest request) {
        if(request==null || request.size()<1 || request.size()>100) throw bad("同步批次须为1至100柜");
@@ -26,21 +26,24 @@ public class AppointmentWorkSyncService {
        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("预约需求扫描中变化，请从首批重试");
+        // 保留全部历史与稳定键顺序；逐柜更新立即进入本次工作集，不跨事务缓存。
+        var known=workingSet();
        int created=0; var ids=new HashSet<String>();
        for(var source:page.containers()) {
            validateSource(source);
            if(!ids.add(source.containerBookingId())) throw unavailable("来源响应重复柜次");
-            var existing=store.container(source.containerBookingId());
+            var existing=known.values().stream().filter(s->source.containerBookingId().equals(s.getContainerBookingId())
+                || s.getContainers().stream().anyMatch(c->source.containerBookingId().equals(c.containerBookingId()))).toList();
            if(existing.isEmpty()) {
                if(source.establishedFlow()==Flow.UNKNOWN) {
-                    var route=create(source,Type.ROUTE,null,Flow.UNKNOWN); route.setFollowupDate(LocalDate.now(route.getTimezone()==null?REMINDER_ZONE:zone(route.getTimezone()))); store.insert(route); created++;
-                } else created+=generate(source,source.establishedFlow(),existing,SYSTEM).size();
+                    var route=create(source,Type.ROUTE,null,Flow.UNKNOWN); route.setFollowupDate(LocalDate.now(route.getTimezone()==null?REMINDER_ZONE:zone(route.getTimezone()))); store.insert(route); known.put(route.getWorkItemKey(),route); created++;
+                } else created+=generate(source,source.establishedFlow(),existing,SYSTEM,known).size();
            } else {
                var activeFlow=existing.stream().filter(s->s.getType()!=Type.ROUTE && !s.isRetired()).map(AppointmentWorkState::getFlow).findFirst().orElse(null);
-                if(activeFlow!=null) created+=generate(source,activeFlow,existing,SYSTEM).size();
+                if(activeFlow!=null) created+=generate(source,activeFlow,existing,SYSTEM,known).size();
                else for(var route:existing) if(route.getType()==Type.ROUTE && !route.isRetired()) {
                    if(source.establishedFlow()!=Flow.UNKNOWN) {
-                        var generated=generate(source,source.establishedFlow(),existing,SYSTEM); created+=generated.size();
+                        var generated=generate(source,source.establishedFlow(),existing,SYSTEM,known); created+=generated.size();
                        route.setFlow(source.establishedFlow()); route.setRetired(true); route.setRetirementReason("ROUTE_CONFIRMED"); route.setReplacementKeys(generated); store.save(route);
                    } else if(!Objects.equals(route.getSourceVersion(),source.sourceVersion())) {
                        route.setSourceVersion(source.sourceVersion()); route.setContainerNo(source.containerNo()); route.setTimezone(source.timezone()); store.save(route);
@@ -76,15 +79,25 @@ public class AppointmentWorkSyncService {
            if(requirement.timezone()!=null) zone(requirement.timezone());
        }
    }
-    /** Caller holds work guard and validates the route against current original facts. */
+    /** 命令入口仍在调用方守卫事务中即时加载工作集。 */
    List<String> generate(ContainerSource source,Flow flow,List<AppointmentWorkState> existing,Actor actor) {
+        return generate(source,flow,existing,actor,workingSet());
+    }
+    /** 与持久层全量读取保持稳定键顺序，并保留退役事项及原读取上限。 */
+    private Map<String,AppointmentWorkState> workingSet() {
+        var known=new TreeMap<String,AppointmentWorkState>();
+        store.all().forEach(state->known.put(state.getWorkItemKey(),state));
+        return known;
+    }
+    /** 同页各柜共享最新状态，新增、跨柜关联和退役仍逐项持久化。 */
+    private List<String> generate(ContainerSource source,Flow flow,List<AppointmentWorkState> existing,Actor actor,
+                                  Map<String,AppointmentWorkState> known) {
        validateSource(source);
        if(flow==Flow.UNKNOWN) throw conflict("真实送货路线尚未明确");
        var requirements=gateway.requirements(source,flow);
        source=new ContainerSource(source.containerBookingId(),source.containerNo(),source.departedAt(),source.departedLocalAt(),source.sailedEvidenceLocalAt(),source.sourceVersion(),source.establishedFlow(),source.timezone(),requirements);
        validateSource(source);
        if(requirements.isEmpty()) throw conflict("真实送货需求尚未明确，不能生成空清单");
-        var known=store.all().stream().collect(Collectors.toMap(AppointmentWorkState::getWorkItemKey,Function.identity(),(a,b)->a));
        var created=new ArrayList<AppointmentWorkState>();
        for(var type:flow==Flow.TRANSFER?List.of(Type.PICKUP,Type.WAREHOUSE_RECEIPT,Type.EMPTY_RETURN):List.of(Type.PICKUP,Type.EMPTY_RETURN)) add(source,type,null,flow,known,created);
        for(var requirement:source.requirements()) {
@@ -12,6 +12,39 @@ import static com.baiyue.lastleg.appointmentwork.AppointmentWorkModels.*;
 import static com.baiyue.lastleg.appointmentwork.AppointmentWorkFactGateway.*;

 class AppointmentWorkSyncServiceTest {
+    /** 百柜同页只读三次完整工作集；新路线、既有历史及重复同步都不丢失。 */
+    @Test void hundredContainersReuseOneGenerationWorkingSetAndKeepUnknownRoutesAndHistory() {
+        var f=new WorkTestFixture();
+        var historical=f.state("old",Type.DELIVERY,Flow.DIRECT,"old-demand");
+        historical.setRetired(true);historical.setRetirementReason("BATCH_REPLACED");
+        historical.setReplacementKeys(List.of("later-work"));f.add(historical);
+        var sourceRows=java.util.stream.IntStream.rangeClosed(1,100).mapToObj(index->{
+            String id=Integer.toString(index);
+            var flow=index<=50?Flow.UNKNOWN:Flow.TRANSFER;
+            var requirements=flow==Flow.UNKNOWN?List.<Requirement>of():List.of(new Requirement("INBOUND:"+id,null,null,
+                "FBA","海外仓","ONT8",null,List.of(),List.of(),List.of()));
+            return new ContainerSource(id,"CNT"+id,null,LocalDateTime.parse("2026-10-01T10:00:00"),null,"v1",flow,null,requirements);
+        }).toList();
+        when(f.gateway.discover(null,100)).thenReturn(new SourcePage(sourceRows,null,true,"scope"));
+        clearInvocations(f.mapper);
+        assertEquals(300,f.sync.synchronize(new SyncRequest(null,100)).created());
+        verify(f.mapper,times(3)).all(AppointmentWorkStore.MAX_ITEMS+1);
+        var states=f.store.all();assertEquals(301,states.size());
+        assertEquals(50,states.stream().filter(s->s.getType()==Type.ROUTE && s.getFlow()==Flow.UNKNOWN).count());
+        assertTrue(f.store.get(historical.getWorkItemKey()).isRetired());
+        assertEquals(List.of("later-work"),f.store.get(historical.getWorkItemKey()).getReplacementKeys());
+        var before=new TreeMap<>(f.rows);int historyCount=f.history.size();
+        clearInvocations(f.mapper);
+        assertEquals(0,f.sync.synchronize(new SyncRequest(null,100)).created());
+        verify(f.mapper,times(3)).all(AppointmentWorkStore.MAX_ITEMS+1);
+        assertEquals(before.keySet(),f.rows.keySet());
+        before.forEach((key,row)->{
+            assertEquals(row.getVersion(),f.rows.get(key).getVersion());
+            assertEquals(row.getStateJson(),f.rows.get(key).getStateJson());
+        });
+        assertEquals(historyCount,f.history.size());
+        assertDoesNotThrow(f.store::requireInitialized);
+    }
    /** 同一实物批次拆成客户事项时，整体事项须有完整去向，运输仍保持唯一。 */
    @Test void batchWideDeliveryIsReplacedByItsExplicitCustomerRequirements() {
        var f=new WorkTestFixture();String batch="FBA_APPOINTMENT:1";
@@ -66,6 +99,7 @@ class AppointmentWorkSyncServiceTest {
        f.sync.synchronize(new SyncRequest(null,10));
        assertEquals(keys,f.store.all().stream().filter(s->s.getType()==Type.DELIVERY && !s.isRetired()).map(AppointmentWorkState::getWorkItemKey).sorted().toList());
    }
+    /** 同页移除主柜后，后柜沿最新对象更新；旧批次恢复也不得覆盖版本或丢客户关联。 */
    @Test void removingPrivateCargoDetachesOldContainerAndRestoresDemandWithoutRetiringOtherCargo() {
        var f=new WorkTestFixture();
        var source1=new SourceReference("C1","ORDER:1","CUSTOMER:1");var source2=new SourceReference("C2","ORDER:2","CUSTOMER:2");
@@ -88,6 +122,21 @@ class AppointmentWorkSyncServiceTest {
        second=new ContainerSource("2","CNT2",null,LocalDateTime.parse("2026-10-01T10:00:00"),null,"v1",Flow.TRANSFER,null,List.of(root2));
        when(f.gateway.discover(null,10)).thenReturn(new SourcePage(List.of(first,second),null,true,"scope"));f.sync.synchronize(new SyncRequest(null,10));
        assertEquals(2,f.store.all().stream().filter(s->s.getType()==Type.DELIVERY && !s.isRetired() && s.getBatchKey()==null).count());
+        var retired=f.store.get(kept.getWorkItemKey());assertTrue(retired.isRetired());
+        int retiredVersion=retired.getVersion();
+        first=new ContainerSource("1","CNT1",null,LocalDateTime.parse("2026-10-01T10:00:00"),null,"v1",Flow.TRANSFER,null,List.of(batch));
+        second=new ContainerSource("2","CNT2",null,LocalDateTime.parse("2026-10-01T10:00:00"),null,"v1",Flow.TRANSFER,null,List.of(batch));
+        when(f.gateway.discover(null,10)).thenReturn(new SourcePage(List.of(first,second),null,true,"scope"));
+        clearInvocations(f.mapper);
+        f.sync.synchronize(new SyncRequest(null,10));
+        verify(f.mapper,times(3)).all(AppointmentWorkStore.MAX_ITEMS+1);
+        var restored=f.store.get(kept.getWorkItemKey());assertFalse(restored.isRetired());
+        assertTrue(restored.getVersion()>retiredVersion);
+        assertEquals(Set.of("1","2"),new HashSet<>(restored.getContainers().stream().map(ContainerReference::containerBookingId).toList()));
+        assertEquals(List.of(source1,source2),restored.getSources());
+        assertEquals(1,f.store.all().stream().filter(s->s.getType()==Type.DELIVERY && !s.isRetired()).count());
+        assertTrue(f.history.stream().anyMatch(h->"SOURCE_REMOVED".equals(h.getAction())));
+        assertTrue(f.history.stream().anyMatch(h->"REPLACED".equals(h.getAction())));
    }
    @Test void removedPrivateBatchCannotBeReplacedByAnotherCustomersUnrelatedDemand() {
        var f=new WorkTestFixture();var oldSource=new SourceReference("C1","ORDER:1","CUSTOMER:1");