diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopAnalyticsProperties.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopAnalyticsProperties.java index adebbd356..c78c36466 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopAnalyticsProperties.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopAnalyticsProperties.java @@ -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 warehouses = new LinkedHashMap<>(); @Data diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopInventoryCollector.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopInventoryCollector.java index b63a1c2b2..7b166a30d 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopInventoryCollector.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopInventoryCollector.java @@ -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"); 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 f045b98f6..0a9986582 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 @@ -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().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().eq("id",taskId).set("routing_manifest_json",null)); } @@ -210,6 +215,7 @@ public class SelfopTaskService { tasks.update(null,new UpdateWrapper().eq("id",taskId).set("routing_manifest_json",null)); } task.setNextRetryAtUtc("DEAD".equals(task.getStatus())?null:now.plusSeconds(Math.min(3600,5L<().eq("id",taskId).set("next_retry_at_utc",null)); } @Transactional(noRollbackFor=BusinessSessionStoppedException.class) public SelfopSyncTask retry(long id,String requestKey,String operator,String reason) { diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopInventoryCollectorTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopInventoryCollectorTest.java index d4cee11cf..eabe2f65c 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopInventoryCollectorTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopInventoryCollectorTest.java @@ -39,12 +39,36 @@ class SelfopInventoryCollectorTest { Consumer 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(); + 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); 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 3920dea7b..2571e9c0d 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 @@ -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() { 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 a0753b047..f1d89dfb2 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 @@ -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 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);