fix(lastleg): pace master data pages without holding collection leases
This commit is contained in:
+2
-2
@@ -37,8 +37,8 @@ public class SelfopMasterDataCollector {
|
|||||||
SelfopCollectionPolicy.requireBusiness(task);SelfopCollectionPolicy.requireBusiness(run);
|
SelfopCollectionPolicy.requireBusiness(task);SelfopCollectionPolicy.requireBusiness(run);
|
||||||
require("MASTER_DATA".equals(task.getTaskType())&&lease.account(),"SELFOP_MASTER_TASK_REQUIRED");
|
require("MASTER_DATA".equals(task.getTaskType())&&lease.account(),"SELFOP_MASTER_TASK_REQUIRED");
|
||||||
var manifest=task.getRoutingManifestJson()==null?new Manifest():read(task.getRoutingManifestJson(),Manifest.class);
|
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.
|
// Master-data endpoints need slower paging than outbound. Yield every page so waiting never owns the account lease.
|
||||||
for(int slice=0;slice<5;slice++) {
|
{
|
||||||
leases.renew(lease);boolean owners=!manifest.isOwnersComplete();int size=owners?1000:100;
|
leases.renew(lease);boolean owners=!manifest.isOwnersComplete();int size=owners?1000:100;
|
||||||
var context=SelfopRequestContext.forTask(task,run,config);
|
var context=SelfopRequestContext.forTask(task,run,config);
|
||||||
LingxingAnalyticsClient.Page<?> page=owners?client.customers(context,manifest.nextPage,size):client.products(context,manifest.nextPage,size);
|
LingxingAnalyticsClient.Page<?> page=owners?client.customers(context,manifest.nextPage,size):client.products(context,manifest.nextPage,size);
|
||||||
|
|||||||
+2
-1
@@ -188,7 +188,8 @@ public class SelfopTaskService {
|
|||||||
lockBusinessAccount(taskId);leases.verify(lease);
|
lockBusinessAccount(taskId);leases.verify(lease);
|
||||||
guardBeforeTaskLock(taskId);
|
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);
|
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)
|
@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,String code,boolean retryable,SelfopLeaseService.Lease lease) {
|
||||||
|
|||||||
+12
-2
@@ -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(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);
|
when(data.productCount(20L,"A")).thenReturn(2L);
|
||||||
collector.collect(task,run,lease);
|
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());
|
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(){
|
@Test void countMismatchCannotPublish(){
|
||||||
when(client.products(any(),eq(1),eq(100))).thenReturn(new LingxingAnalyticsClient.Page<>(List.of(product("C","SKU")),1,100,1L,1));
|
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());
|
assertEquals("SELFOP_MASTER_COUNT_MISMATCH",assertThrows(IllegalStateException.class,()->collector.collect(task,run,lease)).getMessage());
|
||||||
assertNotEquals("PUBLISHED",run.getStatus());
|
assertNotEquals("PUBLISHED",run.getStatus());
|
||||||
}
|
}
|
||||||
@@ -63,7 +66,14 @@ class SelfopMasterDataCollectorTest {
|
|||||||
}
|
}
|
||||||
when(data.productCount(20L,"A")).thenReturn(600L);
|
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);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());
|
verify(data,times(600)).insertProduct(anyLong(),eq(20L),eq("A"),any(),any());
|
||||||
}
|
}
|
||||||
@Test void changedTotalRejectsBeforePersistence(){
|
@Test void changedTotalRejectsBeforePersistence(){
|
||||||
|
|||||||
+10
@@ -326,6 +326,16 @@ class SelfopTaskServiceTest {
|
|||||||
verifyNoInteractions(tasks, runs, scopes, accounts);
|
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() {
|
@Test void budgetYieldDoesNotBecomeFailedAttemptOrClearApplyBarrier() {
|
||||||
var task = task("RUNNING");
|
var task = task("RUNNING");
|
||||||
task.setAttemptCount(99);
|
task.setAttemptCount(99);
|
||||||
|
|||||||
Reference in New Issue
Block a user