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 268f7e451..5f4e039b5 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 @@ -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(); @@ -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) 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 a84290c40..b590682eb 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 @@ -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>(); + 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.>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(); - for(var state:call.>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.>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(); + 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(); } } }