java-project/baiyue-firstleg
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore(lastleg): diagnose work fact synchronization phases

Browse Source
This commit is contained in:
ouyangxq
2026-10-17 11:37:57 +08:00
parent 2971d6f7f0
commit 662dead053
4 changed files with 180 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lastleg-service/src/test/java/com/baiyue/lastleg/appointmentwork/AppointmentWorkFactAdapterTest.java
+87
View File
@@ -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() {
lastleg-service/src/test/java/com/baiyue/lastleg/appointmentwork/AppointmentWorkSyncServiceTest.java
+31
View File
@@ -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(); }