fix(selfop): isolate inventory paging and bound unstable scan retries

2026-10-14 09:36:52 +08:00
parent 0aa188445b
commit 6024faa9aa
6 changed files with 108 additions and 6 deletions
@@ -53,6 +53,8 @@ public class SelfopAnalyticsProperties {
    private int leaseSeconds = 180;
    private int maxAttempts = 5;
    private int pageSize = 100;
+    /** Inventory supports larger pages independently of the outbound list limit. */
+    private int inventoryPageSize = 1000;
    private int applyBatchSize = 100;
    private Map<String, Warehouse> warehouses = new LinkedHashMap<>();
    @Data
@@ -25,12 +25,14 @@ public class SelfopInventoryCollector {
    }
    public boolean collect(SelfopSyncTask task,SelfopSyncRun run,SelfopLeaseService.Lease lease){
        SelfopCollectionPolicy.requireBusiness(task);SelfopCollectionPolicy.requireBusiness(run);
+        int pageSize=config.getInventoryPageSize();
+        require(pageSize>=1&&pageSize<=1000,"SELFOP_INVENTORY_PAGE_SIZE_INVALID");
        var manifest=task.getRoutingManifestJson()==null?new SelfopRoutingManifest():read(task.getRoutingManifestJson(),SelfopRoutingManifest.class);
        var budget=new SelfopBudget(config.getRequestBudget(),config.getTimeBudgetSeconds());
        while(!manifest.isDirectoryComplete()){
            if(!budget.take()){taskService.yield(task.getId(),run.getId(),lease);return false;}
-            leases.renew(lease);var page=client.inventory(SelfopRequestContext.forTask(task,run,config),task.getWarehouseCode(),manifest.getNextPage(),config.getPageSize());
-            require(page.current()==manifest.getNextPage()&&page.size()==config.getPageSize(),"SELFOP_PAGE_MISMATCH");
+            leases.renew(lease);var page=client.inventory(SelfopRequestContext.forTask(task,run,config),task.getWarehouseCode(),manifest.getNextPage(),pageSize);
+            require(page.current()==manifest.getNextPage()&&page.size()==pageSize,"SELFOP_PAGE_MISMATCH");
            require(page.total()!=null&&page.pages()!=null,"SELFOP_PAGE_METADATA_MISSING");
            if(manifest.getTotal()==null){manifest.setTotal(page.total());manifest.setPages(page.pages());}
            require(Objects.equals(manifest.getTotal(),page.total())&&Objects.equals(manifest.getPages(),page.pages()),"SELFOP_PAGE_TOTAL_CHANGED");
@@ -198,10 +198,15 @@ public class SelfopTaskService {
        run.setErrorMessage("自营同步失败，需核对运行记录");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"));
-        if(task.getBusinessRootTaskId()!=null&&continuousBusinessRoot(tasks.selectById(task.getBusinessRootTaskId())))retryable=false;
+        boolean inventory=task.getScopeId()!=null&&"INVENTORY".equals(scopes.selectById(task.getScopeId()).getDataFamily());
+        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.
+            retryable=retryable&&inventory&&Set.of("SELFOP_STOCK_DUPLICATE","SELFOP_STOCK_COUNT_MISMATCH","SELFOP_PAGE_TOTAL_CHANGED").contains(code);
+        }
        task.setStatus(retryable&&failures<config.getMaxAttempts()?"RETRY_WAIT":"DEAD");
        if("APPLYING".equals(task.getApplyState())) task.setApplyState("BLOCKED");
-        if(task.getScopeId()!=null&&"INVENTORY".equals(scopes.selectById(task.getScopeId()).getDataFamily())) {
+        if(inventory) {
            task.setRoutingManifestJson(null);
            tasks.update(null,new UpdateWrapper<SelfopSyncTask>().eq("id",taskId).set("routing_manifest_json",null));
        }
@@ -210,6 +215,7 @@ public class SelfopTaskService {
            tasks.update(null,new UpdateWrapper<SelfopSyncTask>().eq("id",taskId).set("routing_manifest_json",null));
        }
        task.setNextRetryAtUtc("DEAD".equals(task.getStatus())?null:now.plusSeconds(Math.min(3600,5L<<Math.min(9,(int)failures))));task.setUpdatedAt(now);tasks.updateById(task);
+        if("DEAD".equals(task.getStatus()))tasks.update(null,new UpdateWrapper<SelfopSyncTask>().eq("id",taskId).set("next_retry_at_utc",null));
    }
    @Transactional(noRollbackFor=BusinessSessionStoppedException.class)
    public SelfopSyncTask retry(long id,String requestKey,String operator,String reason) {
@@ -39,12 +39,36 @@ class SelfopInventoryCollectorTest {
            Consumer<TransactionStatus> callback = invocation.getArgument(0);
            callback.accept(mock(TransactionStatus.class)); return null;
        }).when(tx).executeWithoutResult(any());
-        config.setCollectionMode("BUSINESS"); config.setEnvironment("production"); config.setPageSize(2); config.setRequestBudget(10); config.setTimeBudgetSeconds(60);
+        config.setCollectionMode("BUSINESS"); config.setEnvironment("production"); config.setInventoryPageSize(2); config.setRequestBudget(10); config.setTimeBudgetSeconds(60);
        collector = new SelfopInventoryCollector(client, stages, tasks, runs, leases, taskService, config, tx);
        task = new SelfopSyncTask(); task.setExecutionPurpose("BUSINESS"); task.setTaskNo("LXCOL-20261012-00000001"); task.setId(20L); task.setSourceAccount("demo"); task.setWarehouseCode("A");
        run = new SelfopSyncRun(); run.setExecutionPurpose("BUSINESS"); run.setAttemptNo(1); run.setTaskId(task.getId()); run.setSourceAccount("demo"); run.setFenceToken(lease.fence()); run.setId(10L); run.setTaskId(20L); run.setStatus("FETCHING");
        run.setScopeSnapshotJson("{\"quantityUnit\":\"PCS\"}"); run.setFetchStartedAtUtc(utcNow());
    }
+    @Test void defaultInventoryPageSizeCollects1001RowsInTwoRequestsWithoutChangingOutboundSize() {
+        var defaults=new SelfopAnalyticsProperties();
+        assertEquals(100,defaults.getPageSize());assertEquals(1000,defaults.getInventoryPageSize());
+        config.setInventoryPageSize(defaults.getInventoryPageSize());
+        var first=new ArrayList<SelfopPayloads.Stock>();
+        for(int i=0;i<1000;i++)first.add(stock("sku-"+i,BigDecimal.ONE));
+        when(client.inventory(any(SelfopRequestContext.class),eq("A"),eq(1),eq(1000)))
+            .thenReturn(new LingxingAnalyticsClient.Page<>(first,1,1000,1001L,2));
+        when(client.inventory(any(SelfopRequestContext.class),eq("A"),eq(2),eq(1000)))
+            .thenReturn(new LingxingAnalyticsClient.Page<>(List.of(stock("last",BigDecimal.ZERO)),2,1000,1001L,2));
+        assertTrue(collector.collect(task,run,lease));
+        assertEquals(100,config.getPageSize());assertEquals(1001L,run.getUniqueObjectCount());
+        assertEquals("COMPLETE",run.getCoverageStatus());assertEquals(2,run.getPageCount());
+        verify(client,times(2)).inventory(any(SelfopRequestContext.class),eq("A"),anyInt(),eq(1000));
+        verify(stages,times(1001)).insert(any(SelfopSyncStage.class));
+    }
+    @Test void invalidInventoryPageSizeRejectsBeforeAnyRequestOrStageWrite() {
+        for(int size:new int[]{0,-1,1001}){
+            config.setInventoryPageSize(size);
+            assertEquals("SELFOP_INVENTORY_PAGE_SIZE_INVALID",assertThrows(IllegalStateException.class,
+                ()->collector.collect(task,run,lease)).getMessage());
+        }
+        verifyNoInteractions(client,stages,leases);
+    }
    @Test void lateInventoryResponseCannotWriteStagesOrReviveStoppedAttempt() {
        when(client.inventory(any(SelfopRequestContext.class),eq("A"),eq(1),eq(2))).thenReturn(page(1,1,1,stock("late",BigDecimal.ONE)));
        when(taskService.checkpoint(task,run,lease)).thenThrow(new SelfopTaskService.BusinessSessionStoppedException("BUSINESS_SESSION_DEADLINE"));
@@ -103,6 +127,25 @@ class SelfopInventoryCollectorTest {
        assertFalse(read(task.getRoutingManifestJson(), SelfopRoutingManifest.class).isDirectoryComplete());
        assertIncompleteAttemptWithPersistedProgress();
    }
+    @Test void freshAttemptAfterDuplicateReReadsAllPagesAndKeepsFailedRowsOutOfItsBatch() {
+        when(client.inventory(any(SelfopRequestContext.class),eq("A"),eq(1),eq(2)))
+            .thenReturn(page(1,3,2,stock("a",BigDecimal.ONE),stock("b",BigDecimal.ZERO)));
+        when(client.inventory(any(SelfopRequestContext.class),eq("A"),eq(2),eq(2)))
+            .thenReturn(page(2,3,2,stock("a",BigDecimal.ONE)),page(2,3,2,stock("c",BigDecimal.TEN)));
+        assertThrows(IllegalStateException.class,()->collector.collect(task,run,lease));
+        assertNull(run.getFetchFinishedAtUtc());
+        // SelfopTaskService.fail/begin clear the failed cursor and allocate a fresh run.
+        task.setRoutingManifestJson(null);
+        var next=new SelfopSyncRun();next.setId(11L);next.setTaskId(task.getId());next.setAttemptNo(2);
+        next.setExecutionPurpose("BUSINESS");next.setSourceAccount("demo");next.setStatus("FETCHING");
+        next.setFetchStartedAtUtc(utcNow());next.setScopeSnapshotJson(run.getScopeSnapshotJson());
+        assertTrue(collector.collect(task,next,lease));
+        verify(client,times(2)).inventory(any(SelfopRequestContext.class),eq("A"),eq(1),eq(2));
+        var captured=ArgumentCaptor.forClass(SelfopSyncStage.class);verify(stages,times(5)).insert(captured.capture());
+        assertEquals(3,captured.getAllValues().stream().filter(s->Objects.equals(s.getStageBatchId(),11L)).count());
+        assertEquals(3L,next.getSourceTotalCount());assertEquals(3L,next.getUniqueObjectCount());
+        assertEquals("COMPLETE",next.getCoverageStatus());assertNull(run.getCoverageStatus());
+    }
    @Test void leakedWarehouseIsRejectedWithoutCompletingRun() {
        var leaked = new SelfopPayloads.Stock("demo", "customer", "B", "sku", 0, null,
            BigDecimal.ONE, BigDecimal.ONE, BigDecimal.ZERO, BigDecimal.ZERO, null);
@@ -50,7 +50,7 @@ class SelfopInventoryRecoveryTest {
        var sales = new SelfopSyncScope(); sales.setId(30L); sales.setDataFamily("OUTBOUND"); sales.setPublishedRunId(80L);
        when(locks.lockTask(1L)).thenReturn(task); when(locks.lockRun(20L)).thenReturn(original);
        when(locks.lockScope(30L)).thenReturn(sales); when(scopes.selectById(10L)).thenReturn(inventory); when(scopes.selectOne(any())).thenReturn(sales);
-        config.setCollectionMode("BUSINESS"); config.setEnvironment("production"); config.setRequestBudget(10); config.setTimeBudgetSeconds(60); config.setPageSize(2);
+        config.setCollectionMode("BUSINESS"); config.setEnvironment("production"); config.setRequestBudget(10); config.setTimeBudgetSeconds(60); config.setInventoryPageSize(2);
    }

    @Test void budgetYieldResumesSameAttemptWithFixedReferencesAndOriginalObservationStart() {
@@ -468,6 +468,55 @@ class SelfopTaskServiceTest {
        assertThat(root.getStatus()).isEqualTo("DEAD");assertThat(root.getNextRetryAtUtc()).isNull();
        assertThat(run.getStatus()).isEqualTo("FAILED");assertThat(run.getErrorCode()).isEqualTo("LINGXING_WMS_RATE_LIMITED");
    }
+    @Test void continuousInventoryScanFailureRetriesFromPageOneWithinFailureBudget(){
+        for(String code:List.of("SELFOP_STOCK_DUPLICATE","SELFOP_STOCK_COUNT_MISMATCH","SELFOP_PAGE_TOTAL_CHANGED")) {
+            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);
+            assertThat(root.getStatus()).isEqualTo("RETRY_WAIT");
+            assertThat(root.getNextRetryAtUtc()).isBetween(before.plusSeconds(10),utcNow().plusSeconds(10));
+            assertThat(root.getRoutingManifestJson()).isNull();
+            assertThat(failed.getStatus()).isEqualTo("FAILED");assertThat(failed.getErrorCode()).isEqualTo(code);
+            assertThat(root.getTaskType()).isEqualTo("SCHEDULED");
+            assertThat(root.getPlannedAtUtc()).isEqualTo(LocalDateTime.of(2026,10,14,0,0));
+            var fresh=service.begin(1L,lease);
+            assertThat(fresh.getId()).isNotEqualTo(failed.getId());
+            assertThat(fresh.getCoverageStatus()).isEqualTo("UNKNOWN");
+            assertThat(fresh.getInventoryDailySampleDate()).isNull();
+            assertThat(root.getRoutingManifestJson()).isNull();
+        }
+    }
+    @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);
+        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")));
+    }
+    @Test void continuousInventoryProtectionAndIdentityErrorsNeverAutomaticallyRetry(){
+        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);
+            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);
+        assertThat(root.getStatus()).isEqualTo("DEAD");
+    }
+    private SelfopSyncTask continuousInventoryRoot(){
+        var root=boundedRoot(0,utcNow().plusMinutes(3));config.setContinuousBusinessCollectionEnabled(true);
+        root.setTaskType("SCHEDULED");root.setStatus("RUNNING");root.setApplyState("NOT_APPLICABLE");
+        root.setRequestedBy(SelfopCollectionPolicy.CONTINUOUS_PLANNER);root.setScopeHash("b".repeat(64));
+        root.setPlannedAtUtc(LocalDateTime.of(2026,10,14,0,0));
+        root.setRequestKey("continuous:inventory:WH-A:2026-10-14T00:00:"+"b".repeat(12));
+        root.setQueryScopeJson(json(Map.of("dataFamily","INVENTORY","businessTimezone","Asia/Shanghai","sourceAccount","synthetic-account","warehouseCode","WH-A")));
+        root.setBusinessBudgetJson(com.baiyue.lastleg.selfoperated.analytics.request.SelfopBusinessBudget.freezeContinuous(utcNow()));
+        when(scopes.selectById(10L)).thenReturn(scope("INVENTORY"));
+        return root;
+    }
    @Test void egressStopReasonSurvivesExecutorFailureHandling() {
        var root=boundedRoot(7,utcNow().plusMinutes(3));
        var budget=read(root.getBusinessBudgetJson(),com.baiyue.lastleg.selfoperated.analytics.request.SelfopBusinessBudget.class);