chore(lastleg): diagnose work fact synchronization phases

This commit is contained in:
2026-10-17 11:37:57 +08:00
parent 2971d6f7f0
commit 662dead053
4 changed files with 180 additions and 9 deletions
@@ -10,6 +10,7 @@ import static com.baiyue.lastleg.appointmentwork.AppointmentWorkPolicy.*;
@Component
public class AppointmentWorkFactAdapter implements AppointmentWorkFactGateway {
private static final org.slf4j.Logger LOG=org.slf4j.LoggerFactory.getLogger(AppointmentWorkFactAdapter.class);
private final AppointmentWorkFirstlegClient firstleg;
private final List<BusinessProvider> providers;
private final ObjectProvider<OwnerDirectory> directory;
@@ -38,7 +39,10 @@ public class AppointmentWorkFactAdapter implements AppointmentWorkFactGateway {
return response.data();
} catch(RuntimeException error) { throw unavailable("柜次权威事实不可用",error); }
}
/** 原顺序读取全部事实;慢调用和失败仅附加无业务内容的局部统计。 */
public Map<String,Fact> facts(List<AppointmentWorkState> states) {
var timing=new FactTiming();boolean success=false;
try {
var result=new LinkedHashMap<String,Fact>();
var remote=new ArrayList<FactRequest>();
var groups=new LinkedHashMap<BusinessProvider,List<FactRequest>>();
@@ -53,8 +57,8 @@ public class AppointmentWorkFactAdapter implements AppointmentWorkFactGateway {
}
}
for(int start=0; start<remote.size(); start+=200) {
var requests=remote.subList(start,Math.min(start+200,remote.size())); requireToken();
try { var response=firstleg.facts(token,requests);
var requests=remote.subList(start,Math.min(start+200,remote.size())); timing.stage="FIRSTLEG"; requireToken();
try { var response=timing.measure(timing.firstleg,requests.size(),()->firstleg.facts(token,requests));
if(response==null || !response.successful()) throw unavailable("提柜及还空事实读取失败");
append(result,requests,response.data());
} catch(RuntimeException error) { throw unavailable("Firstleg工作事实不可用",error); }
@@ -62,11 +66,55 @@ public class AppointmentWorkFactAdapter implements AppointmentWorkFactGateway {
groups.forEach((provider,requests)->{
for(int start=0; start<requests.size(); start+=200) {
var batch=requests.subList(start,Math.min(start+200,requests.size()));
try { append(result,batch,provider.read(batch)); }
var statistics=timing.provider(provider);
try { append(result,batch,timing.measure(statistics,batch.size(),()->provider.read(batch))); }
catch(RuntimeException error) { throw unavailable("预约/收货/运输事实不可用",error); }
}
});
return result;
success=true;return result;
} finally { timing.finish(success); }
}
/** 一次 facts 调用的计时,不保存原来源或返回值,不跨调用缓存。 */
private static final class FactTiming {
private final long started=System.nanoTime();
private final CallStatistics firstleg=new CallStatistics();
private final Map<String,CallStatistics> providers=new LinkedHashMap<>();
private String stage="PREPARE";
/** 固定诊断桶不使用代理名、业务参数或异常文本。 */
private CallStatistics provider(BusinessProvider provider) {
String name=org.springframework.util.ClassUtils.getUserClass(provider).getSimpleName();
name=switch(name) {
case "AppointmentWorkReceivingProvider","AppointmentWorkDeliveryProvider","AppointmentWorkTransportProvider","AppointmentWorkDirectTransportProvider" -> name;
default -> "OTHER";
};
stage=name;
return providers.computeIfAbsent(name,ignored->new CallStatistics());
}
/** 只统计真正尝试的提供端调用;校验及组装耗时保留在 totalMs 中。 */
private <T> T measure(CallStatistics statistics,int requests,java.util.function.Supplier<T> operation) {
statistics.calls++;statistics.requests+=requests;long begin=System.nanoTime();
try { return operation.get(); }
finally { statistics.elapsed+=System.nanoTime()-begin; }
}
/** 成功慢调用或任意失败只输出一次汇总,日志故障不得遮蔽原异常。 */
private void finish(boolean success) {
long totalMs=java.util.concurrent.TimeUnit.NANOSECONDS.toMillis(System.nanoTime()-started);
if(success && totalMs<250) return;
try {
var summaries=new LinkedHashMap<String,String>();
providers.forEach((name,value)->summaries.put(name,value.summary()));
LOG.warn("Appointment work facts diagnostic: totalMs={}, firstlegCalls={}, firstlegRequests={}, firstlegMs={}, providers={}, failedStage={}",
totalMs,firstleg.calls,firstleg.requests,firstleg.millis(),summaries,success?"NONE":stage);
} catch(RuntimeException diagnosticFailure) { /* 仅忽略诊断故障,原事实错误仍沿原路径抛出。 */ }
}
}
/** 仅包含数量和单调时钟耗时的累加器。 */
private static final class CallStatistics {
private int calls,requests;private long elapsed;
/** 输出整毫秒,未计入网络之后的事实完整性校验。 */
private long millis() { return java.util.concurrent.TimeUnit.NANOSECONDS.toMillis(elapsed); }
/** 固定字段格式便于日志汇总,不包含任何请求或响应内容。 */
private String summary() { return "calls="+calls+",requests="+requests+",ms="+millis(); }
}
/** 每次生成建立独立读取器;没有实际需求读取时不引入原业务依赖。 */
public RequirementReader openRequirements() {
@@ -69,10 +69,11 @@ public class AppointmentWorkSyncService {
var facts=gateway.facts(transportItems);
timing.next(SyncPhase.RECONCILE);
reconcileTransport(all,facts);
timing.next(SyncPhase.ACTIVE_FACTS);
timing.next(SyncPhase.ACTIVE_STORE);
all=new ArrayList<>(store.all());
var activeItems=all.stream().filter(s->!s.isRetired()).toList();
timing.activeFactsCount=activeItems.size();
timing.next(SyncPhase.ACTIVE_FACTS);
facts=gateway.facts(activeItems);
timing.next(SyncPhase.OBSERVE_CHECKPOINT);
observe(all,facts,SYSTEM);
@@ -84,7 +85,7 @@ public class AppointmentWorkSyncService {
} finally { timing.finish(success); }
}
/** 固定阶段名用于慢同步定位,不包含来源、游标或异常业务内容。 */
private enum SyncPhase { GUARD, DISCOVER, GENERATE, TRANSPORT_FACTS, RECONCILE, ACTIVE_FACTS, OBSERVE_CHECKPOINT }
private enum SyncPhase { GUARD, DISCOVER, GENERATE, TRANSPORT_FACTS, RECONCILE, ACTIVE_STORE, ACTIVE_FACTS, OBSERVE_CHECKPOINT }
/** 仅统计本次方法执行,不含事务代理提交,不能当作精确锁持有时间。 */
private static final class SyncTiming {
private final long started=System.nanoTime();
@@ -102,9 +103,13 @@ public class AppointmentWorkSyncService {
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());
try {
// 旧 activeFactsMs 仍包含工作表读取、解析及适配;新增明细不能改变旧指标口径。
long activeMs=java.util.concurrent.TimeUnit.NANOSECONDS.toMillis(elapsed[SyncPhase.ACTIVE_STORE.ordinal()]+elapsed[SyncPhase.ACTIVE_FACTS.ordinal()]);
LOG.warn("Appointment work sync slow: totalMs={}, guardMs={}, discoverMs={}, generateMs={}, transportFactsMs={}, reconcileMs={}, activeFactsMs={}, activeStoreMs={}, activeAdapterMs={}, observeCheckpointMs={}, pageContainers={}, transportFactsCount={}, activeFactsCount={}, complete={}, failedStage={}",
totalMs,millis(SyncPhase.GUARD),millis(SyncPhase.DISCOVER),millis(SyncPhase.GENERATE),millis(SyncPhase.TRANSPORT_FACTS),millis(SyncPhase.RECONCILE),activeMs,millis(SyncPhase.ACTIVE_STORE),millis(SyncPhase.ACTIVE_FACTS),millis(SyncPhase.OBSERVE_CHECKPOINT),
pageContainers,transportFactsCount,activeFactsCount,complete,success?"NONE":phase.name());
} catch(RuntimeException diagnosticFailure) { /* 诊断失败不能替换原业务异常或阻断提交。 */ }
}
/** 阶段仅输出整数毫秒,不携带业务标识。 */
private long millis(SyncPhase value) { return java.util.concurrent.TimeUnit.NANOSECONDS.toMillis(elapsed[value.ordinal()]); }
@@ -4,11 +4,98 @@ import org.junit.jupiter.api.Test;
import org.springframework.web.server.ResponseStatusException;
import java.util.*;
import static org.junit.jupiter.api.Assertions.*;
import static org.mockito.Mockito.*;
import static com.baiyue.lastleg.appointmentwork.AppointmentWorkModels.*;
import static com.baiyue.lastleg.appointmentwork.AppointmentWorkFactGateway.*;
import static com.baiyue.lastleg.appointmentwork.WorkTestFixture.*;
class AppointmentWorkFactAdapterTest {
/** 慢成功仅追加聚合诊断,原二百项批量、调用次序和事实结果保持一致。 */
@Test void slowSuccessSummarizesOriginalBatchesWithoutBusinessContent() {
var f=new DiagnosticFixture();
var states=new ArrayList<AppointmentWorkState>();
for(int i=0;i<401;i++) states.add(f.work.state("private-pickup-"+i,Type.PICKUP,Flow.TRANSFER,null));
for(int i=0;i<201;i++) states.add(f.work.state("private-receipt-"+i,Type.WAREHOUSE_RECEIPT,Flow.TRANSFER,null));
when(f.firstleg.facts(anyString(),anyList())).thenAnswer(c->{
var batch=c.<List<FactRequest>>getArgument(1);f.remoteSizes.add(batch.size());
if(f.remoteSizes.size()==1) Thread.sleep(270);
return com.baiyue.common.core.api.ApiResponse.success(batch.stream().map(DiagnosticFixture::factFor).toList());
});
try(var log=new DiagnosticLog()) {
var result=f.adapter.facts(states);
assertEquals(List.of(200,200,1),f.remoteSizes);assertEquals(List.of(200,1),f.providerSizes);
assertEquals(states.stream().map(AppointmentWorkState::getWorkItemKey).toList(),new ArrayList<>(result.keySet()));
for(var state:states) assertEquals(fact(state,null,null,ExecutionStatus.PENDING,null),result.get(state.getWorkItemKey()));
assertEquals(1,log.events().size());var event=log.events().get(0);var text=event.getFormattedMessage();
assertTrue(text.contains("firstlegCalls=3, firstlegRequests=401"));
assertTrue(text.contains("AppointmentWorkReceivingProvider=calls=2,requests=201,ms="));
assertTrue(text.contains("failedStage=NONE"));assertFalse(text.contains("private-"));assertNull(event.getThrowableProxy());
verify(f.firstleg,times(3)).facts(anyString(),anyList());verify(f.provider,times(2)).read(anyList());
}
}
/** 快速失败也输出一次当前阶段;保留原包装异常及其原始原因,不重试。 */
@Test void failedRemoteAndProviderCallsKeepOriginalCauseAndStopAtSameBatch() {
for(boolean remote:List.of(true,false)) {
var f=new DiagnosticFixture();var failure=new IllegalStateException("private-payload-customer-token");
var state=f.work.state("private-container",remote?Type.PICKUP:Type.WAREHOUSE_RECEIPT,Flow.TRANSFER,null);
if(remote) when(f.firstleg.facts(anyString(),anyList())).thenThrow(failure);
else when(f.provider.read(anyList())).thenThrow(failure);
try(var log=new DiagnosticLog()) {
var thrown=assertThrows(ResponseStatusException.class,()->f.adapter.facts(List.of(state)));
assertEquals(503,thrown.getStatusCode().value());assertSame(failure,thrown.getCause());
assertEquals(remote?"Firstleg工作事实不可用":"预约/收货/运输事实不可用",thrown.getReason());
assertEquals(1,log.events().size());var event=log.events().get(0);var text=event.getFormattedMessage();
assertTrue(text.contains("failedStage="+(remote?"FIRSTLEG":"AppointmentWorkReceivingProvider")));
assertTrue(text.contains(remote?"firstlegCalls=1, firstlegRequests=1":"firstlegCalls=0, firstlegRequests=0"));
if(!remote) assertTrue(text.contains("AppointmentWorkReceivingProvider=calls=1,requests=1,ms="));
assertFalse(text.contains("private-"));assertNull(event.getThrowableProxy());
verify(f.firstleg,times(remote?1:0)).facts(anyString(),anyList());verify(f.provider,times(remote?0:1)).read(anyList());
}
}
}
/** 输入分组失败不调用原域,失败诊断仍不得泄露异常中的业务类型内容。 */
@Test void preparationFailureHasNoProviderAttempts() {
var f=new DiagnosticFixture();var state=f.work.state("private-container",Type.TRANSPORT,Flow.TRANSFER,null);
try(var log=new DiagnosticLog()) {
assertThrows(ResponseStatusException.class,()->f.adapter.facts(List.of(state)));
assertEquals(1,log.events().size());var text=log.events().get(0).getFormattedMessage();
assertTrue(text.contains("firstlegCalls=0, firstlegRequests=0"));assertTrue(text.contains("providers={}"));
assertTrue(text.contains("failedStage=PREPARE"));assertFalse(text.contains("private-"));
verifyNoInteractions(f.firstleg);verify(f.provider,never()).read(anyList());
}
}
private static final class DiagnosticFixture {
final WorkTestFixture work=new WorkTestFixture();
final AppointmentWorkFirstlegClient firstleg=mock(AppointmentWorkFirstlegClient.class);
final com.baiyue.lastleg.inboundplan.work.AppointmentWorkReceivingProvider provider=mock(com.baiyue.lastleg.inboundplan.work.AppointmentWorkReceivingProvider.class);
final List<Integer> remoteSizes=new ArrayList<>(),providerSizes=new ArrayList<>();
final AppointmentWorkFactAdapter adapter;
/** 使用真实提供端类型和原分组契约,便于核对固定诊断桶。 */
@SuppressWarnings("unchecked")
DiagnosticFixture() {
var providers=(org.springframework.beans.factory.ObjectProvider<BusinessProvider>)mock(org.springframework.beans.factory.ObjectProvider.class);
when(providers.orderedStream()).thenAnswer(c->java.util.stream.Stream.of(provider));
when(provider.supports(Type.WAREHOUSE_RECEIPT,Flow.TRANSFER)).thenReturn(true);
when(provider.read(anyList())).thenAnswer(c->{var batch=c.<List<FactRequest>>getArgument(0);providerSizes.add(batch.size());return batch.stream().map(DiagnosticFixture::factFor).toList();});
adapter=new AppointmentWorkFactAdapter(firstleg,providers,null,null,null,"private-token");
}
/** 保留原测试事实形状,统计不得影响事实指纹或返回顺序。 */
static Fact factFor(FactRequest request) {
var state=new AppointmentWorkState();state.setWorkItemKey(request.workItemKey());
return fact(state,null,null,ExecutionStatus.PENDING,null);
}
}
private static final class DiagnosticLog implements AutoCloseable {
final ch.qos.logback.classic.Logger logger=(ch.qos.logback.classic.Logger)org.slf4j.LoggerFactory.getLogger(AppointmentWorkFactAdapter.class);
final ch.qos.logback.core.read.ListAppender<ch.qos.logback.classic.spi.ILoggingEvent> appender=new ch.qos.logback.core.read.ListAppender<>();
/** 每个用例独立捕获日志,不修改全局日志级别。 */
DiagnosticLog() { appender.start();logger.addAppender(appender); }
/** 返回本用例的聚合日志记录。 */
List<ch.qos.logback.classic.spi.ILoggingEvent> events() { return appender.list; }
/** 无论断言结果都释放日志采集器。 */
public void close() { logger.detachAppender(appender);appender.stop(); }
}
/** 未读取需求的未知路线不依赖提供端,首次读取才开作用域并逐份校验结果。 */
@SuppressWarnings("unchecked")
@Test void requirementReaderIsLazyAndKeepsProviderCompletenessChecks() {
@@ -321,6 +321,35 @@ class AppointmentWorkSyncServiceTest {
}
}
/** 明细拆分仍保持旧活跃阶段总耗时;表读取失败不追加事实请求或发布。 */
@Test void activeStoreAndAdapterTimingsRetainCombinedMetricAndFailureBehavior() {
for(boolean failStore:List.of(false,true)) {
var f=new WorkTestFixture();var reads=new java.util.concurrent.atomic.AtomicInteger();
var factCalls=new java.util.concurrent.atomic.AtomicInteger();var failure=AppointmentWorkPolicy.unavailable("private-store-body");
when(f.mapper.all(anyInt())).thenAnswer(call->{
if(reads.incrementAndGet()==3) {Thread.sleep(270);if(failStore) throw failure;}
return List.of();
});
when(f.gateway.facts(anyList())).thenAnswer(call->{if(factCalls.incrementAndGet()==2) Thread.sleep(25);return Map.of();});
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 {
if(failStore) {assertSame(failure,assertThrows(ResponseStatusException.class,()->f.sync.synchronize(new SyncRequest(null,10))));verify(f.mapper,never()).saveGuard(any());}
else {assertTrue(f.sync.synchronize(new SyncRequest(null,10)).complete());verify(f.mapper,times(1)).saveGuard(any());}
assertEquals(3,reads.get());assertEquals(failStore?1:2,factCalls.get());assertEquals(1,appender.list.size());
var event=appender.list.get(0);var text=event.getFormattedMessage();
long storeMs=metric(text,"activeStoreMs"),adapterMs=metric(text,"activeAdapterMs"),combined=metric(text,"activeFactsMs");
assertTrue(storeMs>=250);assertTrue(combined>=storeMs+adapterMs && combined<=storeMs+adapterMs+1);
if(failStore) assertEquals(0,adapterMs);else assertTrue(adapterMs>=20);
assertTrue(text.contains("failedStage="+(failStore?"ACTIVE_STORE":"NONE")));assertFalse(text.contains("private-"));assertNull(event.getThrowableProxy());
} finally {logger.detachAppender(appender);appender.stop();}
}
}
/** 从固定诊断字段读取整数,只用于确认原聚合指标口径。 */
private static long metric(String text,String name) {
var match=java.util.regex.Pattern.compile(name+"=(\\d+)").matcher(text);assertTrue(match.find());return Long.parseLong(match.group(1));
}
/** 慢事实失败保留原异常,只记录一次固定阶段摘要,不输出异常业务内容。 */
@Test void slowFactFailureLogsOnePhaseSummaryWithoutPublishingOrLeakingFailure() {
var f=new WorkTestFixture();var calls=new java.util.concurrent.atomic.AtomicInteger();
@@ -338,6 +367,8 @@ class AppointmentWorkSyncServiceTest {
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(metric(event.getFormattedMessage(),"activeAdapterMs")>=250);
assertTrue(metric(event.getFormattedMessage(),"activeFactsMs")>=metric(event.getFormattedMessage(),"activeAdapterMs"));
assertTrue(event.getFormattedMessage().contains("transportFactsCount=0, activeFactsCount=0"));
assertFalse(event.getFormattedMessage().contains("private-source-detail"));assertNull(event.getThrowableProxy());
} finally { logger.detachAppender(appender);appender.stop(); }