diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/appointmentwork/AppointmentWorkStore.java b/lastleg-service/src/main/java/com/baiyue/lastleg/appointmentwork/AppointmentWorkStore.java index b1a295229..e2439d69a 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/appointmentwork/AppointmentWorkStore.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/appointmentwork/AppointmentWorkStore.java @@ -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(); diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/appointmentwork/AppointmentWorkSyncService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/appointmentwork/AppointmentWorkSyncService.java index d91ead756..268f7e451 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/appointmentwork/AppointmentWorkSyncService.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/appointmentwork/AppointmentWorkSyncService.java @@ -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(); 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 generate(ContainerSource source,Flow flow,List existing,Actor actor) { + return generate(source,flow,existing,actor,workingSet()); + } + /** 与持久层全量读取保持稳定键顺序，并保留退役事项及原读取上限。 */ + private Map workingSet() { + var known=new TreeMap(); + store.all().forEach(state->known.put(state.getWorkItemKey(),state)); + return known; + } + /** 同页各柜共享最新状态，新增、跨柜关联和退役仍逐项持久化。 */ + private List generate(ContainerSource source,Flow flow,List existing,Actor actor, + Map 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(); 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()) { diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/appointmentwork/AppointmentWorkSyncServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/appointmentwork/AppointmentWorkSyncServiceTest.java index b44348ca7..a84290c40 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/appointmentwork/AppointmentWorkSyncServiceTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/appointmentwork/AppointmentWorkSyncServiceTest.java @@ -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.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");