From 1112d11c80c32d798500c00242bdff1ead312aec Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Wed, 14 Oct 2026 16:56:40 +0800 Subject: [PATCH] fix(lastleg): pace master data pages without holding collection leases --- .../analytics/sync/SelfopMasterDataCollector.java | 4 ++-- .../analytics/sync/SelfopTaskService.java | 3 ++- .../sync/SelfopMasterDataCollectorTest.java | 14 ++++++++++++-- .../analytics/sync/SelfopTaskServiceTest.java | 10 ++++++++++ 4 files changed, 26 insertions(+), 5 deletions(-) diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopMasterDataCollector.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopMasterDataCollector.java index 31a8403c5..3763b4e0d 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopMasterDataCollector.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopMasterDataCollector.java @@ -37,8 +37,8 @@ public class SelfopMasterDataCollector { SelfopCollectionPolicy.requireBusiness(task);SelfopCollectionPolicy.requireBusiness(run); require("MASTER_DATA".equals(task.getTaskType())&&lease.account(),"SELFOP_MASTER_TASK_REQUIRED"); var manifest=task.getRoutingManifestJson()==null?new Manifest():read(task.getRoutingManifestJson(),Manifest.class); - // Re-read priority after at most five pages, without losing the batch or consuming a failed attempt. - for(int slice=0;slice<5;slice++) { + // Master-data endpoints need slower paging than outbound. Yield every page so waiting never owns the account lease. + { leases.renew(lease);boolean owners=!manifest.isOwnersComplete();int size=owners?1000:100; var context=SelfopRequestContext.forTask(task,run,config); LingxingAnalyticsClient.Page page=owners?client.customers(context,manifest.nextPage,size):client.products(context,manifest.nextPage,size); 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 fe2ac214c..8a7a36134 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 @@ -188,7 +188,8 @@ public class SelfopTaskService { lockBusinessAccount(taskId);leases.verify(lease); guardBeforeTaskLock(taskId); 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.setStatus("YIELDED");run.setFinishedAtUtc(now);run.setUpdatedAt(now);runs.updateById(run); - task.setStatus("RETRY_WAIT");task.setNextRetryAtUtc(now.plusSeconds(1));task.setUpdatedAt(now);tasks.updateById(task); + // Persist the master-data pacing across restarts; ordinary sales/inventory scheduling stays unchanged. + 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) { diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopMasterDataCollectorTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopMasterDataCollectorTest.java index 9d9fc56c6..e42d7e31a 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopMasterDataCollectorTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopMasterDataCollectorTest.java @@ -39,11 +39,14 @@ class SelfopMasterDataCollectorTest { when(client.products(any(),eq(1),eq(100))).thenReturn(new LingxingAnalyticsClient.Page<>(List.of(product("C","SKU"),product("D","SKU")),1,100,2L,1)); when(data.productCount(20L,"A")).thenReturn(2L); collector.collect(task,run,lease); + verify(client,never()).products(any(),anyInt(),anyInt()); + collector.collect(task,run,lease); assertEquals("PUBLISHED",run.getStatus());assertEquals("SUCCEEDED",task.getStatus());assertEquals(3L,run.getUniqueObjectCount()); - verify(data,times(2)).insertProduct(anyLong(),eq(20L),eq("A"),any(),any());verify(service,never()).yield(anyLong(),anyLong(),any()); + verify(data,times(2)).insertProduct(anyLong(),eq(20L),eq("A"),any(),any());verify(service,times(1)).yield(anyLong(),anyLong(),any()); } @Test void countMismatchCannotPublish(){ when(client.products(any(),eq(1),eq(100))).thenReturn(new LingxingAnalyticsClient.Page<>(List.of(product("C","SKU")),1,100,1L,1)); + collector.collect(task,run,lease); assertEquals("SELFOP_MASTER_COUNT_MISMATCH",assertThrows(IllegalStateException.class,()->collector.collect(task,run,lease)).getMessage()); assertNotEquals("PUBLISHED",run.getStatus()); } @@ -63,7 +66,14 @@ class SelfopMasterDataCollectorTest { } when(data.productCount(20L,"A")).thenReturn(600L); collector.collect(task,run,lease);verify(service).yield(10L,20L,lease);assertNotEquals("PUBLISHED",run.getStatus()); - collector.collect(task,run,lease);assertEquals("PUBLISHED",run.getStatus());verify(client,times(1)).customers(any(),eq(1),eq(1000)); + verify(client,never()).products(any(),anyInt(),anyInt()); + for(int page=1;page<=6;page++){ + collector.collect(task,run,lease); + verify(client,times(page)).products(any(),anyInt(),anyInt()); + if(page<6)assertNotEquals("PUBLISHED",run.getStatus()); + } + assertEquals("PUBLISHED",run.getStatus());verify(client,times(1)).customers(any(),eq(1),eq(1000)); + verify(service,times(6)).yield(10L,20L,lease); verify(data,times(600)).insertProduct(anyLong(),eq(20L),eq("A"),any(),any()); } @Test void changedTotalRejectsBeforePersistence(){ 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 1814fb3db..925810900 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 @@ -326,6 +326,16 @@ class SelfopTaskServiceTest { verifyNoInteractions(tasks, runs, scopes, accounts); } + @Test void masterDataYieldPersistsFiveSecondsWithoutChangingOrdinaryScheduling() { + var task=task("RUNNING");task.setTaskType("MASTER_DATA"); + var run=run("FETCHING");when(locks.lockTask(1L)).thenReturn(task);when(locks.lockRun(20L)).thenReturn(run); + service.yield(1L,20L,lease); + assertThat(task.getNextRetryAtUtc()).isEqualTo(task.getUpdatedAt().plusSeconds(5)); + assertThat(run.getStatus()).isEqualTo("YIELDED"); + task.setTaskType("BACKFILL");task.setStatus("RUNNING");run.setStatus("FETCHING"); + service.yield(1L,20L,lease); + assertThat(task.getNextRetryAtUtc()).isEqualTo(task.getUpdatedAt().plusSeconds(1)); + } @Test void budgetYieldDoesNotBecomeFailedAttemptOrClearApplyBarrier() { var task = task("RUNNING"); task.setAttemptCount(99);