fix(selfop): retry continuous outbound failures with bounded diagnostics

2026-10-16 11:05:37 +08:00
parent a3a5862f89
commit 4e2ea89403
11 changed files with 299 additions and 38 deletions
@@ -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<String,String> 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<String> 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<Throwable,Boolean>());
+        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<String,Object>();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; }
+}
@@ -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<String,Object> progress,String field) {
        Object value=progress.get(field);
@@ -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<String,OutboundFactService.Lease> 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";}
 }
@@ -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<SelfopSyncRun>().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<config.getMaxAttempts()?"RETRY_WAIT":"DEAD");
+        task.setStatus(retryable&&failures<=maxRetries?"RETRY_WAIT":"DEAD");
+        run.setErrorMessage(failure.stored(phase,failures,maxRetries,"RETRY_WAIT".equals(task.getStatus())));runs.updateById(run);
        if("APPLYING".equals(task.getApplyState())) task.setApplyState("BLOCKED");
        if(inventory) {
            task.setRoutingManifestJson(null);
@@ -0,0 +1,51 @@
+package com.baiyue.lastleg.selfoperated.analytics.model;
+
+import java.sql.SQLException;
+import java.util.Map;
+import org.junit.jupiter.api.Test;
+import org.springframework.dao.CannotAcquireLockException;
+import static org.assertj.core.api.Assertions.*;
+import static com.baiyue.lastleg.selfoperated.analytics.model.SelfopSupport.*;
+
+class SelfopFailureTest {
+    @Test void wrappedDatabaseFailureRetainsDiagnosticsWithoutRawSqlOrCredentials() {
+        var failure=SelfopFailure.capture("SELFOP_EXECUTION_FAILED",true,
+            new CannotAcquireLockException("SQL customer=private token=secret",new SQLException("password=secret","40001",1213)));
+        assertThat(failure.code()).isEqualTo("SELFOP_DATABASE_TRANSIENT");assertThat(failure.outboundRetryable()).isTrue();
+        String stored=failure.stored("MERGE",1,3,true);
+        assertThat(stored).hasSizeLessThan(500).doesNotContain("private","secret","password","customer");
+        assertThat(SelfopFailure.publicSummary(failure.code(),stored)).contains("归并出库事实失败","SQLException","SQLState 40001","1213","第 1/3 次自动重试");
+    }
+    @Test void permanentDatabaseErrorsAndProviderProtectionDoNotRetry() {
+        var failure=SelfopFailure.capture("SELFOP_EXECUTION_FAILED",true,new RuntimeException("sql",new SQLException("bad syntax","42000",1064)));
+        assertThat(failure.code()).isEqualTo("SELFOP_DATABASE_FAILED");assertThat(failure.outboundRetryable()).isFalse();
+        var provider=SelfopFailure.capture("INTEGRATION_TRANSPORT_FAILURE",true,new RuntimeException("private response"));
+        assertThat(provider.outboundRetryable()).isFalse();
+        assertThat(SelfopFailure.publicSummary(provider.code(),provider.stored("DIRECTORY",1,3,false)))
+            .contains("响应结果未确认","不自动重试").doesNotContain("private");
+        assertThat(SelfopFailure.coded("LINGXING_WMS_SOURCE_REJECTED",true).outboundRetryable()).isTrue();
+        assertThat(SelfopFailure.coded("LINGXING_WMS_SOURCE_REJECTED",false).outboundRetryable()).isFalse();
+    }
+    @Test void diagnosticPhaseUsesOnlyCurrentAttemptCheckpoint() {
+        var task=new SelfopSyncTask();var run=new SelfopSyncRun();run.setAttemptNo(25);run.setStatus("FETCHING");
+        task.setProgressJson(json(Map.of("attemptNo",25,"phase","DETAIL")));
+        assertThat(SelfopFailure.phase(task,run)).isEqualTo("DETAIL");
+        task.setProgressJson(json(Map.of("attemptNo",24,"phase","PUBLISH")));
+        assertThat(SelfopFailure.phase(task,run)).isEqualTo("FETCHING");
+        task.setProgressJson("private invalid json");assertThat(SelfopFailure.phase(task,run)).isEqualTo("FETCHING");
+        var failure=SelfopFailure.coded("SELFOP_EXECUTION_FAILED",true).atPhase("PUBLISH");
+        assertThat(SelfopFailure.publicSummary(failure.code(),failure.stored("BUILD",1,3,true))).startsWith("发布统计结果失败");
+    }
+    @Test void readSideRejectsUnstructuredMessagesAndIgnoresExtraPayloadFields() {
+        assertThat(SelfopFailure.publicSummary("FAIL","token=secret")).doesNotContain("secret");
+        String stored=json(Map.of("v",1,"phase","PUBLISH","type","password=secret","sqlState","credential",
+            "failures",4,"maxRetries",3,"scheduled",false,"payload","secret"));
+        assertThat(SelfopFailure.publicSummary("FAIL",stored)).contains("发布统计结果失败","最多 3 次").doesNotContain("secret","credential","password");
+        assertThat(SelfopFailure.publicSummary("FAIL","x".repeat(501))).doesNotContain("xxx");
+    }
+    @Test void boundedCauseTraversalHandlesCyclesAndRetainsUnexpectedExceptionType() {
+        var a=new RuntimeException("secret");var b=new IllegalStateException("private");a.initCause(b);b.initCause(a);
+        var failure=SelfopFailure.capture("SELFOP_EXECUTION_FAILED",true,a);
+        assertThat(failure.exceptionType()).isEqualTo("IllegalStateException");assertThat(failure.outboundRetryable()).isTrue();
+    }
+}
@@ -381,6 +381,18 @@ class CollectionQueryServiceTest {
        var result=service.runs(q(null,null,null)).records().get(0);
        assertEquals("SOURCE_ERROR",result.errorCode());assertFalse(result.failureSummary().contains("credential"));
    }
+    @Test void runHistoryReturnsFailurePhaseCauseAndAutomaticRetryDecision() {
+        Bound upper=new Bound(now,100L);when(mapper.runBound(any())).thenReturn(upper);
+        var failure=com.baiyue.lastleg.selfoperated.analytics.model.SelfopFailure.capture("SELFOP_EXECUTION_FAILED",true,
+            new RuntimeException("private SQL",new java.sql.SQLException("secret","40001",1213)));
+        var r=new Attempt(90L,7L,"LXCOL-TASK",25,"FAILED","FETCHING","BUSINESS",now.minusHours(1),null,null,
+            now,null,now,null,"PARTIAL",null,null,failure.code(),failure.stored("DETAIL",1,3,true),now,"A",1L,"hash");
+        when(mapper.runs(any(),eq(upper))).thenReturn(List.of(r));
+        var result=service.runs(q(null,null,null)).records().get(0);
+        assertEquals("SELFOP_DATABASE_TRANSIENT",result.errorCode());
+        assertTrue(result.failureSummary().contains("核验出库数据失败"));assertTrue(result.failureSummary().contains("第 1/3 次自动重试"));
+        assertTrue(result.failureSummary().contains("SQLState 40001"));assertFalse(result.failureSummary().contains("private"));
+    }
    @Test void inventoryFreshnessUsesCompletedObservationAndBarrierBlocksInitialCoverage() {
        Scope s=new Scope(1L,"A","OUTBOUND","hash","Asia/Shanghai","0 0 8,20 * * *",93600,
                now.minusMonths(2),now,now.minusMonths(2),now,0,90L,now.minusDays(8),now,7,null,null);
@@ -138,7 +138,7 @@ class SelfopDiscoveryServiceTest {
        account.setRequestBudget(10); service.collect(task, run, lease);
        verify(client, times(1)).locate(any(SelfopRequestContext.class), eq("target"), eq(1), eq(1)); verify(client, times(1)).locate(any(SelfopRequestContext.class), eq("target"), eq(2), eq(1));
        assertEquals("ROUTED", run.getStatus());
-        verify(taskService, never()).fail(anyLong(), anyLong(), anyString(), anyBoolean(), any());
+        verify(taskService, never()).fail(anyLong(), anyLong(), any(SelfopFailure.class), any());
    }
    @Test void stableDirectoryScansYieldAndResumeBeforeDetailsWithoutRepeatingFirstScan() {
        directoryTwoOrders(); account.setRequestBudget(1);
@@ -10,6 +10,25 @@ import java.time.LocalDateTime;
 import static org.mockito.Mockito.*;

 class SelfopExecutionServiceTest {
+    @Test void unexpectedBusinessFailureReachesDurableRetryHandlerWithSafeCauseMetadata() {
+        var config=new SelfopAnalyticsProperties();config.setCollectionMode("BUSINESS");config.setEnvironment("production");
+        var taskService=mock(SelfopTaskService.class);var leases=mock(SelfopLeaseService.class);var tasks=mock(SelfopSyncTaskMapper.class);
+        var discovery=mock(SelfopDiscoveryService.class);
+        var task=new SelfopSyncTask();task.setId(1L);task.setTaskRole("ACCOUNT_DISCOVERY");task.setDiscoveryScopeId(3L);
+        task.setExecutionPurpose("BUSINESS");task.setStatus("READY");task.setTaskNo("LXCOL-TEST");
+        var run=new SelfopSyncRun();run.setId(2L);var lease=new SelfopLeaseService.Lease(3L,"owner",8L,true);
+        when(tasks.selectById(1L)).thenReturn(task);when(leases.acquireDiscovery(eq(3L),anyString())).thenReturn(lease);
+        when(taskService.begin(1L,lease)).thenReturn(run);
+        doThrow(new org.springframework.dao.CannotAcquireLockException("secret SQL",new java.sql.SQLException("private","40001",1213)))
+            .when(discovery).collect(task,run,lease);
+        var service=new SelfopExecutionService(config,taskService,leases,discovery,mock(SelfopInventoryCollector.class),mock(InventorySnapshotService.class),
+            mock(OutboundFactService.class),mock(OutboundCandidateBuilder.class),mock(SelfopSalesPublisher.class),mock(SelfopSyncScopeMapper.class),tasks,
+            mock(SelfopSyncStageMapper.class),mock(SelfopOutboundOrderMapper.class),mock(SelfopReferenceRecovery.class),mock(SelfopValidationRunner.class),
+            mock(SelfopSyncRunMapper.class),mock(SelfopMasterDataCollector.class));
+        service.execute(1L);
+        verify(taskService).fail(eq(1L),eq(2L),argThat(f->"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);
    }
@@ -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));
@@ -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));
@@ -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<<failures),utcNow().plusSeconds(5L<<failures));
+                assertThat(summary).contains("第 "+failures+"/3 次自动重试");
+                var next=service.begin(1L,accountLease);
+                assertThat(next.getTaskId()).isEqualTo(1L);assertThat(root.getRoutingManifestJson()).isEqualTo(checkpoint);
+            } else {assertThat(root.getNextRetryAtUtc()).isNull();assertThat(summary).contains("自动重试已用尽（最多 3 次）");}
+        }
+        verify(tasks,never()).insert(any(SelfopSyncTask.class));
+    }
+    @Test void continuousOutboundWarehouseFailureAlsoRetriesLocalProcessing(){
+        var root=continuousOutboundRoot();root.setStatus("SUCCEEDED");
+        var child=task("RUNNING");child.setId(2L);child.setBusinessRootTaskId(1L);child.setUpstreamTaskId(1L);child.setApplyState("APPLYING");
+        child.setRequestKey("route:1:WH-A");child.setTaskType("BACKFILL");
+        when(tasks.selectById(2L)).thenReturn(child);when(locks.lockTask(2L)).thenReturn(child);
+        when(scopes.selectById(10L)).thenReturn(scope("OUTBOUND"));
+        var failed=run("BUILDING");failed.setTaskId(2L);when(locks.lockRun(20L)).thenReturn(failed);when(runs.selectCount(any())).thenReturn(1L);
+        service.fail(2L,20L,SelfopFailure.capture("SELFOP_EXECUTION_FAILED",true,
+            new org.springframework.dao.CannotAcquireLockException("private SQL",new java.sql.SQLException("private params","40001",1213))),lease);
+        assertThat(child.getStatus()).isEqualTo("RETRY_WAIT");assertThat(child.getApplyState()).isEqualTo("BLOCKED");
+        assertThat(root.getStatus()).isEqualTo("SUCCEEDED");assertThat(failed.getErrorCode()).isEqualTo("SELFOP_DATABASE_TRANSIENT");
+        assertThat(SelfopFailure.publicSummary(failed.getErrorCode(),failed.getErrorMessage())).contains("SQLState 40001","第 1/3").doesNotContain("private");
+    }
+    @Test void continuousOutboundProtectionAndContractFailuresNeverRetry(){
+        for(String code:List.of("LINGXING_WMS_RATE_LIMITED","LINGXING_WMS_AUTHORIZATION_REJECTED","INTEGRATION_TRANSPORT_FAILURE",
+                "SELFOP_SCOPE_MISMATCH","SELFOP_IDENTITY_INVALID","SELFOP_DATABASE_FAILED")) {
+            var root=continuousOutboundRoot();when(locks.lockRun(20L)).thenReturn(run("FETCHING"));when(runs.selectCount(any())).thenReturn(1L);
+            service.fail(1L,20L,SelfopFailure.coded(code,true),new SelfopLeaseService.Lease(99L,"owner",7L,true));
+            assertThat(root.getStatus()).as(code).isEqualTo("DEAD");assertThat(root.getNextRetryAtUtc()).isNull();
+        }
+        var root=continuousOutboundRoot();when(locks.lockRun(20L)).thenReturn(run("FETCHING"));
+        service.fail(1L,20L,SelfopFailure.coded("SELFOP_EXECUTION_FAILED",false),new SelfopLeaseService.Lease(99L,"owner",7L,true));
+        assertThat(root.getStatus()).isEqualTo("DEAD");
+    }
+    @Test void continuousFailureStopsInsteadOfBecomingAnotherScheduledRetry(){
+        var root=continuousOutboundRoot();
        var run=run("FETCHING");when(locks.lockRun(20L)).thenReturn(run);
-        service.fail(1L,20L,"LINGXING_WMS_RATE_LIMITED",true,new SelfopLeaseService.Lease(99L,"owner",7L,true));
+        service.fail(1L,20L,SelfopFailure.coded("LINGXING_WMS_RATE_LIMITED",true),new SelfopLeaseService.Lease(99L,"owner",7L,true));
        assertThat(root.getStatus()).isEqualTo("DEAD");assertThat(root.getNextRetryAtUtc()).isNull();
        assertThat(run.getStatus()).isEqualTo("FAILED");assertThat(run.getErrorCode()).isEqualTo("LINGXING_WMS_RATE_LIMITED");
    }
@@ -483,7 +534,7 @@ class SelfopTaskServiceTest {
            var root=continuousInventoryRoot();root.setRoutingManifestJson("{\"nextPage\":6}");root.setAttemptCount(99);
            var failed=run("FETCHING");when(locks.lockRun(20L)).thenReturn(failed);
            when(runs.selectCount(any())).thenReturn(1L);config.setMaxAttempts(3);
-            var before=utcNow();service.fail(1L,20L,code,true,lease);
+            var before=utcNow();service.fail(1L,20L,SelfopFailure.coded(code,true),lease);
            assertThat(root.getStatus()).isEqualTo("RETRY_WAIT");
            assertThat(root.getNextRetryAtUtc()).isBetween(before.plusSeconds(10),utcNow().plusSeconds(10));
            assertThat(root.getRoutingManifestJson()).isNull();
@@ -500,7 +551,7 @@ class SelfopTaskServiceTest {
    @Test void continuousInventoryScanFailureStopsAtConfiguredLimitAndClearsPersistedRetryTime(){
        var root=continuousInventoryRoot();root.setNextRetryAtUtc(utcNow().minusMinutes(1));
        when(locks.lockRun(20L)).thenReturn(run("FETCHING"));when(runs.selectCount(any())).thenReturn(3L);config.setMaxAttempts(3);
-        service.fail(1L,20L,"SELFOP_STOCK_DUPLICATE",true,lease);
+        service.fail(1L,20L,SelfopFailure.coded("SELFOP_STOCK_DUPLICATE",true),lease);
        assertThat(root.getStatus()).isEqualTo("DEAD");assertThat(root.getNextRetryAtUtc()).isNull();
        verify(tasks,atLeastOnce()).update(isNull(),argThat((com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper<SelfopSyncTask> 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");
    }