From 4e2ea894039fc2130c5ed78fb3c0451203191c70 Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Fri, 16 Oct 2026 11:05:37 +0800 Subject: [PATCH] fix(selfop): retry continuous outbound failures with bounded diagnostics --- .../analytics/model/SelfopFailure.java | 100 ++++++++++++++++++ .../query/CollectionQueryService.java | 4 +- .../sync/SelfopExecutionService.java | 32 +++--- .../analytics/sync/SelfopTaskService.java | 21 ++-- .../analytics/model/SelfopFailureTest.java | 51 +++++++++ .../query/CollectionQueryServiceTest.java | 12 +++ .../sync/SelfopDiscoveryServiceTest.java | 2 +- .../sync/SelfopExecutionServiceTest.java | 24 ++++- .../sync/SelfopInventoryRecoveryTest.java | 4 +- .../sync/SelfopSchedulePlannerTest.java | 12 +++ .../analytics/sync/SelfopTaskServiceTest.java | 75 ++++++++++--- 11 files changed, 299 insertions(+), 38 deletions(-) create mode 100644 lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopFailure.java create mode 100644 lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopFailureTest.java diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopFailure.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopFailure.java new file mode 100644 index 000000000..e7f840f41 --- /dev/null +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopFailure.java @@ -0,0 +1,100 @@ +package com.baiyue.lastleg.selfoperated.analytics.model; + +import com.fasterxml.jackson.databind.JsonNode; +import java.sql.SQLException; +import java.util.*; +import static com.baiyue.lastleg.selfoperated.analytics.model.SelfopSupport.*; + +/** Persist only diagnostic metadata, never exception messages, SQL, parameters or provider bodies. */ +public record SelfopFailure(String code, boolean retryable, String exceptionType, String sqlState, Integer vendorCode, String executionPhase) { + public static final int OUTBOUND_MAX_RETRIES = 3; + private static final Map PHASES = Map.ofEntries( + Map.entry("DIRECTORY","获取出库目录"),Map.entry("DETAIL","核验出库数据"),Map.entry("ROUTE","分派仓库任务"), + Map.entry("MERGE","归并出库事实"),Map.entry("BUILD","计算统计结果"),Map.entry("PUBLISH","发布统计结果"), + Map.entry("INVENTORY","获取库存"),Map.entry("CUSTOMERS","获取客户资料"),Map.entry("PRODUCTS","获取商品资料"), + Map.entry("FETCHING","获取数据"),Map.entry("BUILDING","计算统计结果"),Map.entry("UNKNOWN","执行采集任务")); + private static final Set OUTBOUND_RETRY_CODES = Set.of("SELFOP_EXECUTION_FAILED","SELFOP_DATABASE_TRANSIENT", + "SELFOP_DIRECTORY_DUPLICATE","SELFOP_DIRECTORY_COUNT_MISMATCH","SELFOP_DIRECTORY_UNSTABLE","SELFOP_PAGE_TOTAL_CHANGED", + "LINGXING_WMS_SOURCE_REJECTED"); + + public static SelfopFailure coded(String code, boolean retryable) { + return new SelfopFailure(code,retryable,null,null,null,null); + } + public static SelfopFailure capture(String code, boolean retryable, Throwable error) { + String state=null;Integer vendor=null;Throwable deepest=error; + var seen=Collections.newSetFromMap(new IdentityHashMap()); + boolean transientDatabase=false,database=false; + for(Throwable cause=error;cause!=null&&seen.size()<12&&seen.add(cause);cause=cause.getCause()) { + deepest=cause; + if(cause instanceof SQLException sql) { + database=true;state=sql.getSQLState();vendor=sql.getErrorCode(); + transientDatabase |= "40001".equals(state)||vendor==1213||vendor==1205; + } + if(cause instanceof org.springframework.dao.DataAccessException)database=true; + if(cause instanceof org.springframework.dao.TransientDataAccessException + ||cause instanceof java.sql.SQLTransientException||cause instanceof java.sql.SQLRecoverableException)transientDatabase=true; + } + if("SELFOP_EXECUTION_FAILED".equals(code)&&database) { + code=transientDatabase?"SELFOP_DATABASE_TRANSIENT":"SELFOP_DATABASE_FAILED"; + retryable=transientDatabase; + } + return new SelfopFailure(code,retryable,token(deepest.getClass().getSimpleName(),"[A-Za-z][A-Za-z0-9_$]{0,79}"), + token(state,"[A-Z0-9]{5}"),vendor,null); + } + public boolean outboundRetryable() { return retryable&&OUTBOUND_RETRY_CODES.contains(code); } + public SelfopFailure atPhase(String phase) { + return new SelfopFailure(code,retryable,exceptionType,sqlState,vendorCode,phase!=null&&PHASES.containsKey(phase)?phase:null); + } + + public String stored(String phase,long failures,int maxRetries,boolean scheduled) { + if(executionPhase!=null)phase=executionPhase; + var data=new LinkedHashMap();data.put("v",1);data.put("phase",PHASES.containsKey(phase)?phase:"UNKNOWN"); + data.put("type",exceptionType);data.put("sqlState",sqlState);data.put("vendorCode",vendorCode); + data.put("failures",failures);data.put("maxRetries",maxRetries);data.put("scheduled",scheduled); + return json(data); + } + public static String phase(SelfopSyncTask task,SelfopSyncRun run) { + if(task.getProgressJson()!=null)try { + var p=read(task.getProgressJson(),JsonNode.class); + if(p.path("attemptNo").asInt(-1)==Objects.requireNonNullElse(run.getAttemptNo(),-2)) { + String phase=p.path("phase").asText();if(PHASES.containsKey(phase))return phase; + } + }catch(IllegalArgumentException ignored) { } + return PHASES.containsKey(run.getStatus())?run.getStatus():"UNKNOWN"; + } + public static String publicSummary(String code,String stored) { + String generic="采集尝试失败，请根据错误码查看受控诊断记录"; + if(stored==null)return null; + if(stored.length()>500)return generic; + try { + JsonNode p=read(stored,JsonNode.class); + if(p.path("v").asInt()!=1||!PHASES.containsKey(p.path("phase").asText()) + ||!p.path("failures").canConvertToLong()||p.path("failures").asLong()<1 + ||!p.path("maxRetries").canConvertToInt()||p.path("maxRetries").asInt()<0 + ||!p.path("scheduled").isBoolean())return generic; + String cause=switch(code==null?"":code) { + case "SELFOP_DATABASE_TRANSIENT" -> "数据库临时锁冲突或连接故障"; + case "SELFOP_DATABASE_FAILED" -> "数据库操作失败"; + case "SELFOP_DIRECTORY_DUPLICATE","SELFOP_DIRECTORY_COUNT_MISMATCH","SELFOP_DIRECTORY_UNSTABLE","SELFOP_PAGE_TOTAL_CHANGED" -> "目录完整性校验未通过"; + case "LINGXING_WMS_RATE_LIMITED","LINGXING_WMS_HTTP_RATE_LIMITED","LINGXING_WMS_SOURCE_RATE_LIMITED" -> "来源接口限流"; + case "INTEGRATION_TRANSPORT_FAILURE" -> "接口传输失败，响应结果未确认"; + case "LINGXING_WMS_AUTHORIZATION_REJECTED" -> "来源接口鉴权失败"; + case "LINGXING_WMS_SOURCE_REJECTED" -> "来源接口拒绝请求"; + default -> "执行异常"; + }; + StringBuilder result=new StringBuilder(PHASES.get(p.path("phase").asText())).append("失败：").append(cause); + String type=token(p.path("type").asText(null),"[A-Za-z][A-Za-z0-9_$]{0,79}"); + if(type!=null)result.append("；异常类型 ").append(type); + String state=token(p.path("sqlState").asText(null),"[A-Z0-9]{5}"); + if(state!=null)result.append("；SQLState ").append(state); + if(p.path("vendorCode").isIntegralNumber())result.append("；数据库错误码 ").append(p.path("vendorCode").intValue()); + long failures=p.path("failures").longValue();int max=p.path("maxRetries").intValue(); + result.append("；累计失败 ").append(failures).append(" 次"); + if(p.path("scheduled").booleanValue())result.append("，已安排第 ").append(failures).append("/").append(max).append(" 次自动重试"); + else if(failures>max)result.append("，自动重试已用尽（最多 ").append(max).append(" 次），需人工检查"); + else result.append("，该错误不自动重试，需核对错误码及账号保护状态"); + return result.toString(); + }catch(IllegalArgumentException ignored) { return generic; } + } + private static String token(String value,String pattern) { return value!=null&&value.matches(pattern)?value:null; } +} diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionQueryService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionQueryService.java index 03844c131..846a243cf 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionQueryService.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionQueryService.java @@ -327,9 +327,7 @@ public class CollectionQueryService { private String safeMessage(String code,String error) { if("BUSINESS_SESSION_HTTP_LIMIT".equals(code))return "本次正式采集会话的请求预算已耗尽，根任务及关联子任务已停止，不能恢复原任务"; if("BUSINESS_SESSION_DEADLINE".equals(code))return "该任务曾因旧采集截止时间停止，需核对原检查点和账户保护状态后受控恢复"; - if(error==null)return null; - // Errors are already sanitized by the write side; avoid exposing arbitrary payloads here. - return "采集尝试失败，请根据错误码查看受控诊断记录"; + return com.baiyue.lastleg.selfoperated.analytics.model.SelfopFailure.publicSummary(code,error); } private Instant progressInstant(Map progress,String field) { Object value=progress.get(field); diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopExecutionService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopExecutionService.java index b29965ba9..482600e2b 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopExecutionService.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopExecutionService.java @@ -12,6 +12,7 @@ import static com.baiyue.lastleg.selfoperated.analytics.model.SelfopSupport.*; @Service public class SelfopExecutionService { + private static final org.slf4j.Logger LOG=org.slf4j.LoggerFactory.getLogger(SelfopExecutionService.class); private final SelfopAnalyticsProperties config; private final SelfopTaskService taskService; private final SelfopLeaseService leases; @@ -59,16 +60,18 @@ public class SelfopExecutionService { } held.putAll(leases.acquireScopes(ids,owner));primary=held.get(task.getScopeId()); } - SelfopSyncRun run=null; + SelfopSyncRun run=null;String executionPhase=null; try{ run=taskService.begin(taskId,primary); task=tasks.selectById(taskId); if("VALIDATION".equals(task.getExecutionPurpose())){validation.run(task,run,primary);return;} - if(!primary.account()&&"OUTBOUND".equals(scopes.selectById(task.getScopeId()).getDataFamily()))recovery.refresh(task,held); + if(!primary.account()&&"OUTBOUND".equals(scopes.selectById(task.getScopeId()).getDataFamily())){executionPhase="MERGE";recovery.refresh(task,held);} if("MASTER_DATA".equals(task.getTaskType()))masterData.collect(task,run,primary); else if(primary.account())discovery.collect(task,run,primary); else if("INVENTORY".equals(scopes.selectById(task.getScopeId()).getDataFamily())){ - if(stock.collect(task,run,primary)){snapshots.build(task.getScopeId(),run.getId(),owner,primary.fence());snapshots.publish(task.getScopeId(),run.getId(),owner,primary.fence());} + executionPhase="INVENTORY"; + if(stock.collect(task,run,primary)){executionPhase="BUILD";snapshots.build(task.getScopeId(),run.getId(),owner,primary.fence()); + executionPhase="PUBLISH";snapshots.publish(task.getScopeId(),run.getId(),owner,primary.fence());} }else{ Map outbound=new TreeMap<>();for(var lease:held.values())outbound.put(scopes.selectById(lease.scopeId()).getWarehouseCode(),new OutboundFactService.Lease(owner,lease.fence())); if(!"REBUILD".equals(task.getTaskType())){ @@ -79,23 +82,28 @@ public class SelfopExecutionService { facts.finish(taskId,run.getId(),ids,outbound); } } - var reused=candidates.reusable(task.getScopeId(),run.getId(),owner,primary.fence()); - if(reused!=null){publisher.reuse(task.getScopeId(),run.getId(),reused,owner,primary.fence());return;} + executionPhase="BUILD";var reused=candidates.reusable(task.getScopeId(),run.getId(),owner,primary.fence()); + if(reused!=null){executionPhase="PUBLISH";publisher.reuse(task.getScopeId(),run.getId(),reused,owner,primary.fence());return;} candidates.build(task.getScopeId(),run.getId(),owner,primary.fence()); - if(config.isSalesPublicationEnabled())publisher.publish(task.getScopeId(),run.getId(),owner,primary.fence()); + if(config.isSalesPublicationEnabled()){executionPhase="PUBLISH";publisher.publish(task.getScopeId(),run.getId(),owner,primary.fence());} else taskService.candidateFinished(taskId,run.getId(),primary); } }catch(RuntimeException e){ if(run!=null&&"BUSINESS".equals(task.getExecutionPurpose())&& e instanceof com.baiyue.lastleg.selfoperated.analytics.request.EgressRejectedException q && q.waitUntilUtc()!=null){ try{taskService.waitForBudget(taskId,run.getId(),q.waitUntilUtc(),primary);}catch(IllegalStateException lostFence){ }return; } - if(run!=null){String code=e instanceof AnalyticsClientException c?c.code():safeCode(e.getMessage()); - boolean retryable=!(e instanceof AnalyticsClientException c)||c.retryable(); + var failure=SelfopFailure.capture(e instanceof AnalyticsClientException c?c.code():safeCode(e.getMessage()), + !(e instanceof AnalyticsClientException c)||c.retryable(),e).atPhase(executionPhase); + // Stack locations and classified metadata only: throwable messages can contain SQL and credentials. + LOG.warn("Selfop execution failed taskNo={} runId={} diagnostic={} locations={}",task.getTaskNo(),run==null?null:run.getId(), + failure,Arrays.stream(e.getStackTrace()).limit(8).toList()); + if(run!=null){ try{ - if("VALIDATION".equals(task.getExecutionPurpose()))validation.fail(task,run,primary,code); - else taskService.fail(taskId,run.getId(),code,retryable,primary); - }catch(IllegalStateException lostFence){/* New lease owner decides recovery. */}} + if("VALIDATION".equals(task.getExecutionPurpose()))validation.fail(task,run,primary,failure.code()); + else taskService.fail(taskId,run.getId(),failure,primary); + }catch(IllegalStateException lostFence){LOG.warn("Selfop failure persistence rejected taskNo={} type={}",task.getTaskNo(),lostFence.getClass().getSimpleName());}} }finally{for(var lease:held.values())leases.release(lease);} } - private static String safeCode(String message){return message!=null&&message.matches("[A-Z][A-Z0-9_]{0,63}")?message:"SELFOP_EXECUTION_FAILED";} + private static String safeCode(String message){return message!=null&&message.length()<=64 + &&message.matches("(SELFOP|BUSINESS|ACCOUNT|VALIDATION|LINGXING|INTEGRATION)_[A-Z0-9_]+")?message:"SELFOP_EXECUTION_FAILED";} } diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskService.java index 8a7a36134..4ff32a8b4 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskService.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskService.java @@ -192,21 +192,28 @@ public class SelfopTaskService { task.setStatus("RETRY_WAIT");task.setNextRetryAtUtc(now.plusSeconds("MASTER_DATA".equals(task.getTaskType())?5:1));task.setUpdatedAt(now);tasks.updateById(task); } @Transactional(noRollbackFor=BusinessSessionStoppedException.class) - public void fail(long taskId,long runId,String code,boolean retryable,SelfopLeaseService.Lease lease) { + public void fail(long taskId,long runId,SelfopFailure failure,SelfopLeaseService.Lease lease) { lockBusinessAccount(taskId);leases.verify(lease); try{guardBeforeTaskLock(taskId);}catch(BusinessSessionStoppedException stopped){return;} - var now=utcNow();var task=locks.lockTask(taskId);var run=locks.lockRun(runId);checkRunLease(task,run,lease);requireBusiness(task);requireBusiness(run);SelfopProgressProjection.freezeRun(task,run);run.setLastPhase(run.getStatus());run.setStatus("FAILED");run.setErrorCode(code); - run.setErrorMessage("自营同步失败，需核对运行记录");run.setFinishedAtUtc(now);run.setUpdatedAt(now);runs.updateById(run); + var now=utcNow();var task=locks.lockTask(taskId);var run=locks.lockRun(runId);checkRunLease(task,run,lease);requireBusiness(task);requireBusiness(run); + String code=failure.code();boolean retryable=failure.retryable();String phase=SelfopFailure.phase(task,run); + SelfopProgressProjection.freezeRun(task,run);run.setLastPhase(run.getStatus());run.setStatus("FAILED");run.setErrorCode(code); + run.setFinishedAtUtc(now);run.setUpdatedAt(now);runs.updateById(run); // A yielded execution is not a failed request; finite failures are counted separately. long failures=runs.selectCount(new QueryWrapper().eq("task_id",taskId).eq("status","FAILED")); boolean inventory=task.getScopeId()!=null&&"INVENTORY".equals(scopes.selectById(task.getScopeId()).getDataFamily()); + int maxRetries=Math.max(0,config.getMaxAttempts()-1); if(task.getBusinessRootTaskId()!=null&&continuousBusinessRoot(tasks.selectById(task.getBusinessRootTaskId()))) { - // A completed response with an unstable inventory scan can be read again from page one. - // Authentication, rate limits, uncertain transport results and other families retain their stop policy. + boolean outbound=!"MASTER_DATA".equals(task.getTaskType()) + &&("ACCOUNT_DISCOVERY".equals(task.getTaskRole())||task.getScopeId()!=null&&"OUTBOUND".equals(scopes.selectById(task.getScopeId()).getDataFamily())); + // Retry the same durable task, never a new daily root that would reset the failure budget. + if(outbound)maxRetries=SelfopFailure.OUTBOUND_MAX_RETRIES; retryable=retryable&&(inventory&&Set.of("SELFOP_STOCK_DUPLICATE","SELFOP_STOCK_COUNT_MISMATCH","SELFOP_PAGE_TOTAL_CHANGED").contains(code) - ||"MASTER_DATA".equals(task.getTaskType())&&Set.of("SELFOP_MASTER_DUPLICATE","SELFOP_MASTER_COUNT_MISMATCH","SELFOP_PAGE_TOTAL_CHANGED").contains(code)); + ||"MASTER_DATA".equals(task.getTaskType())&&Set.of("SELFOP_MASTER_DUPLICATE","SELFOP_MASTER_COUNT_MISMATCH","SELFOP_PAGE_TOTAL_CHANGED").contains(code) + ||outbound&&failure.outboundRetryable()); } - task.setStatus(retryable&&failures"SELFOP_DATABASE_TRANSIENT".equals(f.code())&&f.outboundRetryable()&&"40001".equals(f.sqlState())),eq(lease)); + verify(leases).release(lease); + } @Test void completedDayBuildsAndPublishesImmediatelyWithoutBuildIntervalWait() { var config=new SelfopAnalyticsProperties();config.setCollectionMode("BUSINESS");config.setEnvironment("production");config.setSalesPublicationEnabled(true); var taskService=mock(SelfopTaskService.class);var leases=mock(SelfopLeaseService.class);var tasks=mock(SelfopSyncTaskMapper.class);var scopes=mock(SelfopSyncScopeMapper.class); @@ -24,6 +43,9 @@ class SelfopExecutionServiceTest { service.execute(1L); var order=inOrder(candidates,publisher);order.verify(candidates).build(eq(3L),eq(2L),anyString(),eq(8L));order.verify(publisher).publish(eq(3L),eq(2L),anyString(),eq(8L)); verify(taskService,never()).waitForBudget(anyLong(),anyLong(),any(),any());verifyNoInteractions(runs); + doThrow(new IllegalStateException("private publication payload")).when(publisher).publish(eq(3L),eq(2L),anyString(),eq(8L)); + service.execute(1L); + verify(taskService).fail(eq(1L),eq(2L),argThat(f->"PUBLISH".equals(f.executionPhase())&&"SELFOP_EXECUTION_FAILED".equals(f.code())),eq(lease)); } @Test void unexpectedValidationFailureTerminatesSessionInsteadOfSchedulingRetry() { executeValidationFailure(new IllegalStateException("VALIDATION_STRUCTURE_REJECTED"),"VALIDATION_STRUCTURE_REJECTED"); @@ -51,7 +73,7 @@ class SelfopExecutionServiceTest { recovery,validation,mock(SelfopSyncRunMapper.class),org.mockito.Mockito.mock(SelfopMasterDataCollector.class)); service.execute(1L); verify(validation).fail(task,run,lease,code);verify(leases).release(lease); - verify(taskService,never()).fail(anyLong(),anyLong(),anyString(),anyBoolean(),any()); + verify(taskService,never()).fail(anyLong(),anyLong(),any(SelfopFailure.class),any()); verify(taskService,never()).waitForBudget(anyLong(),anyLong(),any(),any()); verifyNoInteractions(discovery,stock,snapshots,facts,candidates,publisher,recovery); } diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopInventoryRecoveryTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopInventoryRecoveryTest.java index 2571e9c0d..35cc86996 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopInventoryRecoveryTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopInventoryRecoveryTest.java @@ -94,7 +94,7 @@ class SelfopInventoryRecoveryTest { @Test void genuineFailureClearsCursorAndNewAttemptStartsAtPageOneWithItsOwnStageBatch() { original.setFenceToken(lease.fence()); when(runs.selectCount(any())).thenReturn(1L); - service.fail(1L, 20L, "SELFOP_STOCK_DUPLICATE", true, lease); + service.fail(1L, 20L,SelfopFailure.coded("SELFOP_STOCK_DUPLICATE", true), lease); assertEquals("FAILED", original.getStatus()); assertNull(task.getRoutingManifestJson()); assertEquals("RETRY_WAIT", task.getStatus()); // Verifies the final updateById cannot overwrite the explicit SQL-null cursor reset. @@ -122,7 +122,7 @@ class SelfopInventoryRecoveryTest { @Test void exhaustedFailuresRetainOldPublishedPointerAndCannotCreateAnotherAttempt() { original.setFenceToken(lease.fence()); when(runs.selectCount(any())).thenReturn((long) config.getMaxAttempts()); - service.fail(1L, 20L, "SOURCE_TIMEOUT", true, lease); + service.fail(1L, 20L,SelfopFailure.coded("SOURCE_TIMEOUT", true), lease); assertEquals("DEAD", task.getStatus()); assertEquals("FAILED", original.getStatus()); assertEquals(90L, inventory.getPublishedRunId()); assertEquals(11L, original.getBasePublishedRunId()); assertThrows(IllegalStateException.class, () -> service.begin(1L, lease)); diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSchedulePlannerTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSchedulePlannerTest.java index a1191d188..8830c0629 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSchedulePlannerTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSchedulePlannerTest.java @@ -16,6 +16,18 @@ import static org.assertj.core.api.Assertions.*; import static org.mockito.Mockito.*; class SelfopSchedulePlannerTest { + @Test void dailyPlanningNeverResetsRetryBudgetByReplacingWaitingOrExhaustedRoot(){ + var state=continuousState();var now=LocalDateTime.of(2026,10,13,4,0); + planner.planContinuous(now); + var root=state.stream().filter(t->t.getRequestKey().startsWith("continuous:outbound:2026-10-12:")).findFirst().orElseThrow(); + String key=root.getRequestKey();int count=state.size(); + for(String status:List.of("RETRY_WAIT","DEAD")) { + root.setStatus(status);root.setAttemptCount(103); + planner.planContinuous(now.plusMinutes(1)); + assertThat(state).hasSize(count);assertThat(state.stream().filter(t->key.equals(t.getRequestKey()))).containsExactly(root); + assertThat(root.getStatus()).isEqualTo(status);assertThat(root.getAttemptCount()).isEqualTo(103); + } + } @Test void masterCatalogIsOneAccountBatchAndAnUnfinishedBatchSpansDays(){ var state=continuousState(); planner.planContinuous(LocalDateTime.of(2026,10,13,4,0)); diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskServiceTest.java index 925810900..d23368d36 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskServiceTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskServiceTest.java @@ -306,7 +306,7 @@ class SelfopTaskServiceTest { @Test void lostLeaseFailCannotTouchCurrentRunOrTask() { doThrow(new IllegalStateException("SELFOP_FENCE_REJECTED")).when(leases).verify(lease); - assertThatThrownBy(() -> service.fail(1L, 20L, "SOURCE_TEMPORARY", true, lease)) + assertThatThrownBy(() -> service.fail(1L, 20L,SelfopFailure.coded("SOURCE_TEMPORARY", true), lease)) .hasMessage("SELFOP_FENCE_REJECTED"); verifyNoInteractions(locks, tasks, runs, scopes, accounts); } @@ -317,7 +317,7 @@ class SelfopTaskServiceTest { oldRun.setFenceToken(6L); when(locks.lockTask(1L)).thenReturn(task); when(locks.lockRun(20L)).thenReturn(oldRun); - assertThatThrownBy(() -> service.fail(1L, 20L, "SOURCE_TEMPORARY", true, lease)) + assertThatThrownBy(() -> service.fail(1L, 20L,SelfopFailure.coded("SOURCE_TEMPORARY", true), lease)) .hasMessage("SELFOP_RUN_LEASE_MISMATCH"); verify(leases).verify(lease); assertThat(task.getStatus()).isEqualTo("RUNNING"); @@ -370,7 +370,7 @@ class SelfopTaskServiceTest { assertThat(query.getParamNameValuePairs()).containsValue("FAILED").doesNotContainValue("YIELDED"); return 2L; }); - service.fail(1L, 20L, "SOURCE_TEMPORARY", true, lease); + service.fail(1L, 20L,SelfopFailure.coded("SOURCE_TEMPORARY", true), lease); assertThat(task.getStatus()).isEqualTo("RETRY_WAIT"); assertThat(task.getApplyState()).isEqualTo("BLOCKED"); assertThat(run.getStatus()).isEqualTo("FAILED"); @@ -387,7 +387,7 @@ class SelfopTaskServiceTest { when(locks.lockRun(20L)).thenReturn(run("BUILDING")); when(scopes.selectById(10L)).thenReturn(scope("OUTBOUND")); when(runs.selectCount(any())).thenReturn(3L); - service.fail(1L, 20L, "SOURCE_TEMPORARY", true, lease); + service.fail(1L, 20L,SelfopFailure.coded("SOURCE_TEMPORARY", true), lease); assertThat(task.getStatus()).isEqualTo("DEAD"); assertThat(task.getApplyState()).isEqualTo("BLOCKED"); } @@ -397,7 +397,7 @@ class SelfopTaskServiceTest { var attempt=run("FETCHING");attempt.setExecutionPurpose("VALIDATION"); when(locks.lockTask(1L)).thenReturn(child);when(locks.lockRun(20L)).thenReturn(attempt); when(tasks.selectById(1L)).thenReturn(child);when(locks.lockScope(10L)).thenReturn(scope("OUTBOUND")); - assertThatThrownBy(()->service.fail(1L,20L,"VALIDATION_STRUCTURE_REJECTED",true,lease)) + assertThatThrownBy(()->service.fail(1L,20L,SelfopFailure.coded("VALIDATION_STRUCTURE_REJECTED",true),lease)) .hasMessage("SELFOP_BUSINESS_PURPOSE_REQUIRED"); assertThatThrownBy(()->service.retry(1L,"validation-retry","operator","offline retry")) .hasMessage("SELFOP_BUSINESS_PURPOSE_REQUIRED"); @@ -465,7 +465,7 @@ class SelfopTaskServiceTest { assertThat(root.getNextRetryAtUtc()).isEqualTo(until);assertThat(root.getBusinessBudgetJson()).isEqualTo(budget); verify(tasks,never()).update(any(),any()); } - @Test void continuousFailureStopsInsteadOfBecomingAnotherScheduledRetry(){ + private SelfopSyncTask continuousOutboundRoot(){ var root=boundedRoot(0,utcNow().plusMinutes(3));config.setContinuousBusinessCollectionEnabled(true); root.setTaskRole("ACCOUNT_DISCOVERY");root.setTaskType("BACKFILL");root.setScopeId(null);root.setWarehouseCode(null);root.setDiscoveryScopeId(99L); root.setRequestedBy(SelfopCollectionPolicy.CONTINUOUS_PLANNER);root.setScopeHash("a".repeat(64)); @@ -473,8 +473,59 @@ class SelfopTaskServiceTest { root.setQueryFromUtc(LocalDateTime.of(2026,10,11,16,0));root.setQueryUntilUtc(LocalDateTime.of(2026,10,12,16,0)); root.setQueryScopeJson(json(new SelfopDiscoveryService.Query("2026-10-12T00:00","2026-10-13T00:00",null,null,null))); root.setBusinessBudgetJson(com.baiyue.lastleg.selfoperated.analytics.request.SelfopBusinessBudget.freezeContinuous(utcNow())); + var account=new SelfopOutboundDiscoveryScope();account.setId(99L);when(locks.lockDiscovery(99L)).thenReturn(account); + return root; + } + @Test void continuousOutboundRetriesThreeTimesThenStopsWithoutResettingCheckpointOrIdentity(){ + var root=continuousOutboundRoot();root.setAttemptCount(99); + root.setRoutingManifestJson("{\"directoryComplete\":true,\"nextPage\":26,\"scan\":2}"); + String checkpoint=root.getRoutingManifestJson(),key=root.getRequestKey(),budget=root.getBusinessBudgetJson(); + var accountLease=new SelfopLeaseService.Lease(99L,"owner",7L,true); + for(int failures=1;failures<=4;failures++) { + root.setStatus("RUNNING");var failed=run("FETCHING");failed.setAttemptNo(root.getAttemptCount());when(locks.lockRun(20L)).thenReturn(failed); + when(runs.selectCount(any())).thenReturn((long)failures); + var before=utcNow();service.fail(1L,20L,SelfopFailure.coded("SELFOP_EXECUTION_FAILED",true),accountLease); + assertThat(root.getStatus()).isEqualTo(failures<=3?"RETRY_WAIT":"DEAD"); + assertThat(root.getRoutingManifestJson()).isEqualTo(checkpoint);assertThat(root.getRequestKey()).isEqualTo(key); + assertThat(root.getBusinessBudgetJson()).isEqualTo(budget); + String summary=SelfopFailure.publicSummary(failed.getErrorCode(),failed.getErrorMessage()); + if(failures<=3) { + assertThat(root.getNextRetryAtUtc()).isBetween(before.plusSeconds(5L< update)-> update.getSqlSet().contains("next_retry_at_utc"))); @@ -509,11 +560,11 @@ class SelfopTaskServiceTest { for(String code:List.of("LINGXING_WMS_RATE_LIMITED","LINGXING_WMS_AUTHORIZATION_REJECTED", "INTEGRATION_TRANSPORT_FAILURE","SELFOP_STOCK_SCOPE_LEAK","SELFOP_STOCK_IDENTITY_MISSING","SELFOP_EXECUTION_FAILED")) { var root=continuousInventoryRoot();when(locks.lockRun(20L)).thenReturn(run("FETCHING"));when(runs.selectCount(any())).thenReturn(1L); - service.fail(1L,20L,code,true,lease); + service.fail(1L,20L,SelfopFailure.coded(code,true),lease); assertThat(root.getStatus()).as(code).isEqualTo("DEAD");assertThat(root.getNextRetryAtUtc()).isNull(); } var root=continuousInventoryRoot();when(locks.lockRun(20L)).thenReturn(run("FETCHING")); - service.fail(1L,20L,"SELFOP_STOCK_DUPLICATE",false,lease); + service.fail(1L,20L,SelfopFailure.coded("SELFOP_STOCK_DUPLICATE",false),lease); assertThat(root.getStatus()).isEqualTo("DEAD"); } @Test void commandRetriesDeadContinuousInventoryThroughRealTaskServiceAndRecordsAudit(){ @@ -568,7 +619,7 @@ class SelfopTaskServiceTest { var root=boundedRoot(7,utcNow().plusMinutes(3)); var budget=read(root.getBusinessBudgetJson(),com.baiyue.lastleg.selfoperated.analytics.request.SelfopBusinessBudget.class); budget.stopReason="BUSINESS_SESSION_HTTP_LIMIT";root.setBusinessBudgetJson(json(budget));root.setStatus("DEAD"); - service.fail(1L,20L,"SELFOP_EXECUTION_FAILED",true,lease); + service.fail(1L,20L,SelfopFailure.coded("SELFOP_EXECUTION_FAILED",true),lease); verify(runs,never()).updateById(any(SelfopSyncRun.class));verify(tasks,never()).updateById(any(SelfopSyncTask.class)); assertThat(read(root.getBusinessBudgetJson(),com.baiyue.lastleg.selfoperated.analytics.request.SelfopBusinessBudget.class).stopReason).isEqualTo("BUSINESS_SESSION_HTTP_LIMIT"); }