perf(selfop): batch list checkpoints and publish each completed day
This commit is contained in:
-4
@@ -17,10 +17,6 @@ public class OutboundCandidateBuilder {
|
||||
if("FINISH".equals(checkpoint.phase()))transaction.finish(scopeId,runId,owner,fence);
|
||||
}
|
||||
public Long reusable(Long scopeId,Long runId,String owner,long fence) {return transaction.reusable(scopeId,runId,owner,fence);}
|
||||
public static boolean regularBuildDue(SelfopSyncTask task,LocalDateTime previousBuild,LocalDateTime now) {
|
||||
return "REBUILD".equals(task.getTaskType()) || "RECHECK".equals(task.getTaskType()) || task.getAttemptCount()!=null && task.getAttemptCount()>1
|
||||
|| previousBuild==null || !now.isBefore(previousBuild.plusMinutes(30));
|
||||
}
|
||||
public static Coverage contiguous(LocalDateTime initialFrom,List<Coverage> intervals) {
|
||||
if(initialFrom==null)return new Coverage(null,null);LocalDateTime until=initialFrom;
|
||||
for(Coverage interval:intervals.stream().filter(i->i.from()!=null && i.until()!=null && i.from().isBefore(i.until()))
|
||||
|
||||
+51
-1
@@ -93,8 +93,12 @@ public class SelfopDiscoveryService {
|
||||
}
|
||||
}
|
||||
route(task,run,manifest,lease,false);
|
||||
for(var entry:manifest.getObjects().entrySet()) {
|
||||
var entries=new ArrayList<>(manifest.getObjects().entrySet());
|
||||
for(int index=0;index<entries.size();index++) {
|
||||
if(!budget.hasTime()){taskService.yield(task.getId(),run.getId(),lease);return;}
|
||||
int accepted=acceptListBatch(task,run,lease,manifest,entries,index);
|
||||
if(accepted>0){index+=accepted-1;continue;}
|
||||
var entry=entries.get(index);
|
||||
var ref=entry.getValue();
|
||||
if(ref.getRefreshTaskId()!=null){
|
||||
var refresh=tasks.selectById(ref.getRefreshTaskId());require(refresh!=null,"SELFOP_REFRESH_TASK_MISSING");
|
||||
@@ -211,6 +215,52 @@ public class SelfopDiscoveryService {
|
||||
.set("catalog_from_utc",account.getCatalogFromUtc()).set("catalog_until_utc",account.getCatalogUntilUtc()).set("updated_at",now));}
|
||||
});
|
||||
}
|
||||
/** Complete list rows need one bounded transaction/checkpoint per batch, not per order.
|
||||
* Missing data, resumed references and newer revisions still follow the individual resolution path.
|
||||
* Stages and their manifest references commit together; no network call occurs in this transaction. */
|
||||
private int acceptListBatch(SelfopSyncTask task,SelfopSyncRun run,SelfopLeaseService.Lease lease,
|
||||
SelfopRoutingManifest manifest,List<Map.Entry<String,SelfopRoutingManifest.Entry>> entries,int start) {
|
||||
var candidates=new ArrayList<Map.Entry<String,SelfopRoutingManifest.Entry>>();
|
||||
for(int i=start;i<Math.min(entries.size(),start+100);i++) {
|
||||
var e=entries.get(i);var ref=e.getValue();
|
||||
if(ref.getEffectiveStageId()!=null||ref.getRefreshTaskId()!=null||ref.getDirectoryStageId()==null)break;
|
||||
candidates.add(e);
|
||||
}
|
||||
if(candidates.isEmpty())return 0;
|
||||
var directory=new HashMap<Long,SelfopSyncStage>();
|
||||
stages.selectList(new QueryWrapper<SelfopSyncStage>().in("id",candidates.stream().map(e->e.getValue().getDirectoryStageId()).toList()))
|
||||
.forEach(s->directory.put(s.getId(),s));
|
||||
var revised=new HashSet<String>();
|
||||
stages.selectList(new QueryWrapper<SelfopSyncStage>().select("object_key_hash").eq("object_type","OUTBOUND_ORDER")
|
||||
.in("object_key_hash",candidates.stream().map(Map.Entry::getKey).toList()).eq("validation_status","VALID")
|
||||
.gt("observation_sequence",task.getObservationBaseline())).forEach(s->revised.add(s.getObjectKeyHash()));
|
||||
var payloads=new ArrayList<SelfopPayloads.Order>();
|
||||
for(var e:candidates) {
|
||||
var s=directory.get(e.getValue().getDirectoryStageId());
|
||||
if(s==null||!Objects.equals(s.getStageBatchId(),task.getId())||!"PENDING".equals(s.getValidationStatus())||revised.contains(e.getKey()))break;
|
||||
var p=read(s.getPayloadJson(),SelfopPayloads.Order.class);
|
||||
if(p==null||LingxingAnalyticsClient.needsDetail(p))break;
|
||||
require(Objects.equals(p.sourceAccount(),task.getSourceAccount())&&Objects.equals(p.customerCode(),e.getValue().getCustomerCode())
|
||||
&&Objects.equals(p.deliveryNo(),e.getValue().getDeliveryNo()),"SELFOP_DETAIL_IDENTITY_MISMATCH");
|
||||
require(p.sourceStatus()!=null,"SELFOP_DETAIL_STRUCTURE_INVALID");identity(p.warehouseCode(),64);
|
||||
payloads.add(p);
|
||||
}
|
||||
if(payloads.isEmpty())return 0;
|
||||
leases.renew(lease);
|
||||
tx.executeWithoutResult(s->{
|
||||
var checkpoint=taskService.checkpoint(task,run,lease);
|
||||
long first=leases.nextObservations(lease,payloads.size());
|
||||
for(int i=0;i<payloads.size();i++) {
|
||||
var ref=candidates.get(i).getValue();var old=directory.get(ref.getDirectoryStageId());var p=payloads.get(i);
|
||||
var full=stage(task,run,p,old.getFirstPageNo(),first+i,"VALID");
|
||||
full.setId(old.getId());full.setRunId(old.getRunId());full.setCreatedAt(old.getCreatedAt());
|
||||
require(stages.updateById(full)==1,"SELFOP_STAGE_MISSING");
|
||||
ref.setEffectiveStageId(full.getId());ref.setWarehouseCode(p.warehouseCode());
|
||||
}
|
||||
saveManifest(checkpoint.task(),checkpoint.run(),manifest,true);
|
||||
});
|
||||
return payloads.size();
|
||||
}
|
||||
private boolean collectLocated(SelfopSyncTask task,SelfopSyncRun run,SelfopLeaseService.Lease lease,SelfopBudget budget,Query query,SelfopRoutingManifest manifest) {
|
||||
if(manifest.isDirectoryComplete())return true;
|
||||
// Identity relocation consumes the full paginated result before trusting one match.
|
||||
|
||||
-4
@@ -78,10 +78,6 @@ public class SelfopExecutionService {
|
||||
}
|
||||
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;}
|
||||
var previousBuild=runs.selectOne(new QueryWrapper<SelfopSyncRun>().eq("scope_id",task.getScopeId()).isNotNull("build_started_at_utc").orderByDesc("build_started_at_utc").last("LIMIT 1"));
|
||||
if(previousBuild!=null&&!OutboundCandidateBuilder.regularBuildDue(task,previousBuild.getBuildStartedAtUtc(),utcNow())){
|
||||
taskService.waitForBudget(taskId,run.getId(),previousBuild.getBuildStartedAtUtc().plusMinutes(30),primary);return;
|
||||
}
|
||||
candidates.build(task.getScopeId(),run.getId(),owner,primary.fence());
|
||||
if(config.isSalesPublicationEnabled())publisher.publish(task.getScopeId(),run.getId(),owner,primary.fence());
|
||||
else taskService.candidateFinished(taskId,run.getId(),primary);
|
||||
|
||||
+7
-1
@@ -66,9 +66,15 @@ public class SelfopLeaseService {
|
||||
}
|
||||
@Transactional
|
||||
public long nextObservation(Lease lease) {
|
||||
return nextObservations(lease,1);
|
||||
}
|
||||
/** Reserve consecutive observations under the same fence; gaps after a failed batch are harmless. */
|
||||
@Transactional
|
||||
public long nextObservations(Lease lease,int count) {
|
||||
require(count>0&&count<=100,"SELFOP_OBSERVATION_BATCH_INVALID");
|
||||
require(lease.account(),"SELFOP_ACCOUNT_LEASE_REQUIRED"); var s=locks.lockDiscovery(lease.scopeId());
|
||||
check(lease,s.getLeaseOwner(),s.getFenceToken(),s.getLeaseUntilUtc(),utcNow());
|
||||
long seq=s.getNextObservationSequence(); s.setNextObservationSequence(seq+1); s.setUpdatedAt(utcNow());
|
||||
long seq=s.getNextObservationSequence(); s.setNextObservationSequence(Math.addExact(seq,count)); s.setUpdatedAt(utcNow());
|
||||
discoveries.updateById(s); return seq;
|
||||
}
|
||||
public void verify(Lease lease) {
|
||||
|
||||
-7
@@ -173,13 +173,6 @@ class OutboundCandidateBuilderTest {
|
||||
assertNull(OutboundCandidateBuilder.class.getMethod("build",Long.class,Long.class,String.class,long.class).getAnnotation(org.springframework.transaction.annotation.Transactional.class));
|
||||
for(String name:List.of("begin","batch","finish"))assertNotNull(OutboundCandidateTransaction.class.getMethod(name,Long.class,Long.class,String.class,long.class).getAnnotation(org.springframework.transaction.annotation.Transactional.class));
|
||||
}
|
||||
@Test void onlyRoutineFirstCollectionIsCoalesced() {
|
||||
task.setTaskType("FIRST_COLLECTION"); task.setAttemptCount(1); LocalDateTime now=SelfopSupport.utcNow();
|
||||
assertFalse(OutboundCandidateBuilder.regularBuildDue(task,now.minusMinutes(29),now));
|
||||
assertTrue(OutboundCandidateBuilder.regularBuildDue(task,now.minusMinutes(30),now));
|
||||
task.setTaskType("REBUILD"); assertTrue(OutboundCandidateBuilder.regularBuildDue(task,now,now));
|
||||
task.setTaskType("FIRST_COLLECTION"); task.setAttemptCount(2); assertTrue(OutboundCandidateBuilder.regularBuildDue(task,now,now));
|
||||
}
|
||||
@Test void missingWarehouseSummaryPreventsReadyAndOldReadyCannotBypassBuild() {
|
||||
transaction.begin(1L,10L,"worker",2);
|
||||
var row=new SelfopSkuDailySales();row.setCustomerCode("c");row.setSkuCode("sku");row.setBusinessDate(LocalDate.of(2026,10,1));
|
||||
|
||||
+48
@@ -378,6 +378,54 @@ class SelfopDiscoveryServiceTest {
|
||||
assertEquals("ROUTED",run.getStatus());
|
||||
assertEquals(1,read(task.getRoutingManifestJson(),SelfopRoutingManifest.class).getListPayloadVersion());
|
||||
}
|
||||
private void prepareCompleteBatch(int count) {
|
||||
var manifest=new SelfopRoutingManifest();manifest.setListPayloadVersion(1);manifest.setDirectoryComplete(true);manifest.setPages(1);manifest.setTotal((long)count);
|
||||
for(int i=0;i<count;i++) {
|
||||
var payload=detail("batch-"+i,"2026-10-11 12:00:00");String key=hash(List.of("demo","customer",payload.deliveryNo()));
|
||||
var stage=new SelfopSyncStage();stage.setId(1000L+i);stage.setRunId(run.getId());stage.setStageBatchId(task.getId());stage.setObjectKeyHash(key);
|
||||
stage.setValidationStatus("PENDING");stage.setFirstPageNo(1);stage.setPayloadJson(json(payload));stage.setCreatedAt(from);stored.put(stage.getId(),stage);
|
||||
var ref=new SelfopRoutingManifest.Entry();ref.setCustomerCode("customer");ref.setWarehouseCode("A");ref.setDeliveryNo(payload.deliveryNo());ref.setDirectoryStageId(stage.getId());manifest.getObjects().put(key,ref);
|
||||
}
|
||||
task.setRoutingManifestJson(json(manifest));
|
||||
when(stages.selectList(any())).thenAnswer(inv->{
|
||||
QueryWrapper<SelfopSyncStage> q=inv.getArgument(0);String sql=q.getSqlSegment();var values=q.getParamNameValuePairs().values();
|
||||
if(sql.contains("observation_sequence"))return List.of();
|
||||
return stored.values().stream().filter(s->values.contains(s.getId())).toList();
|
||||
});
|
||||
long[] next={101};when(leases.nextObservations(eq(lease),anyInt())).thenAnswer(inv->{long first=next[0];next[0]+=inv.<Integer>getArgument(1);return first;});
|
||||
}
|
||||
@Test void completeListsCommitBoundedBatchesInsteadOfPerOrderManifests() {
|
||||
prepareCompleteBatch(205);service.collect(task,run,lease);
|
||||
verifyNoInteractions(client);verify(leases,times(2)).nextObservations(lease,100);verify(leases).nextObservations(lease,5);
|
||||
verify(tasks,times(4)).updateById(argThat((SelfopSyncTask t)->task.getId().equals(t.getId())));
|
||||
assertEquals(205,stored.values().stream().filter(s->"VALID".equals(s.getValidationStatus())).count());
|
||||
assertEquals(205,stored.values().stream().map(SelfopSyncStage::getObservationSequence).distinct().count());
|
||||
assertEquals(101L,stored.values().stream().mapToLong(SelfopSyncStage::getObservationSequence).min().orElseThrow());
|
||||
assertEquals(305L,stored.values().stream().mapToLong(SelfopSyncStage::getObservationSequence).max().orElseThrow());
|
||||
assertEquals("ROUTED",run.getStatus());assertTrue(children.values().stream().allMatch(c->"READY".equals(c.getStatus())));
|
||||
}
|
||||
@Test void batchYieldResumesOnlyUncommittedRowsWithoutSupplierRequests() {
|
||||
prepareCompleteBatch(205);
|
||||
try(var budgets=mockConstruction(SelfopBudget.class,(b,c)->when(b.hasTime()).thenReturn(true,false))){service.collect(task,run,lease);}
|
||||
assertEquals(100,stored.values().stream().filter(s->"VALID".equals(s.getValidationStatus())).count());
|
||||
assertTrue(children.values().stream().allMatch(c->"WAITING_SOURCE".equals(c.getStatus())));
|
||||
service.collect(task,run,lease);verifyNoInteractions(client);
|
||||
verify(stages,times(205)).updateById(any(SelfopSyncStage.class));assertEquals("ROUTED",run.getStatus());
|
||||
}
|
||||
@Test void stoppedBatchCannotPromoteRowsOrAllocateObservations() {
|
||||
prepareCompleteBatch(100);int[] checkpoints={0};
|
||||
when(taskService.checkpoint(task,run,lease)).thenAnswer(inv->{if(++checkpoints[0]>1)throw new SelfopTaskService.BusinessSessionStoppedException("STOPPED");return new SelfopTaskService.CollectionCheckpoint(task,run);});
|
||||
assertThrows(SelfopTaskService.BusinessSessionStoppedException.class,()->service.collect(task,run,lease));
|
||||
verify(leases,never()).nextObservations(any(),anyInt());verify(stages,never()).updateById(any(SelfopSyncStage.class));
|
||||
assertTrue(stored.values().stream().allMatch(s->"PENDING".equals(s.getValidationStatus())));verifyNoInteractions(client);
|
||||
}
|
||||
@Test void incompleteRowInBatchStillUsesDetailAndPreservesCompleteRows() {
|
||||
prepareCompleteBatch(3);var missing=stored.get(1001L);var p=read(missing.getPayloadJson(),SelfopPayloads.Order.class);
|
||||
missing.setPayloadJson(json(new SelfopPayloads.Order(p.sourceAccount(),p.customerCode(),p.warehouseCode(),p.deliveryNo(),p.sourceNo(),p.sourceStatus(),p.sourceCreatedTime(),p.sourceOutboundTime(),null)));
|
||||
service.collect(task,run,lease);verify(client).detail(any(),eq("A"),eq("customer"),eq("batch-1"));
|
||||
verify(client,never()).detail(any(),eq("A"),eq("customer"),eq("batch-0"));verify(client,never()).detail(any(),eq("A"),eq("customer"),eq("batch-2"));
|
||||
assertTrue(stored.values().stream().allMatch(s->"VALID".equals(s.getValidationStatus())));
|
||||
}
|
||||
@Test void validationPurposeCannotReachAccountRoutingOrFormalCoverage() {
|
||||
task.setExecutionPurpose("VALIDATION");run.setExecutionPurpose("VALIDATION");
|
||||
assertEquals("SELFOP_BUSINESS_PURPOSE_REQUIRED",assertThrows(IllegalStateException.class,()->service.collect(task,run,lease)).getMessage());
|
||||
|
||||
+15
@@ -10,6 +10,21 @@ import java.time.LocalDateTime;
|
||||
import static org.mockito.Mockito.*;
|
||||
|
||||
class SelfopExecutionServiceTest {
|
||||
@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);
|
||||
var task=new SelfopSyncTask();task.setId(1L);task.setExecutionPurpose("BUSINESS");task.setStatus("READY");task.setTaskRole("WAREHOUSE_SYNC");task.setTaskType("BACKFILL");task.setScopeId(3L);task.setWarehouseCode("A");task.setApplyState("APPLIED");task.setAttemptCount(1);task.setRoutingManifestJson("{\"objects\":{}}");
|
||||
var scope=new SelfopSyncScope();scope.setId(3L);scope.setWarehouseCode("A");scope.setDataFamily("OUTBOUND");var run=new SelfopSyncRun();run.setId(2L);
|
||||
var lease=new SelfopLeaseService.Lease(3L,"worker",8L,false);
|
||||
when(tasks.selectById(1L)).thenReturn(task);when(scopes.selectById(3L)).thenReturn(scope);when(scopes.selectOne(any())).thenReturn(scope);
|
||||
when(leases.acquireScopes(any(),anyString())).thenReturn(java.util.Map.of(3L,lease));when(taskService.begin(1L,lease)).thenReturn(run);
|
||||
var candidates=mock(OutboundCandidateBuilder.class);var publisher=mock(SelfopSalesPublisher.class);var runs=mock(SelfopSyncRunMapper.class);
|
||||
when(candidates.reusable(anyLong(),anyLong(),anyString(),anyLong())).thenReturn(null);
|
||||
var service=new SelfopExecutionService(config,taskService,leases,mock(SelfopDiscoveryService.class),mock(SelfopInventoryCollector.class),mock(InventorySnapshotService.class),mock(OutboundFactService.class),candidates,publisher,scopes,tasks,mock(SelfopSyncStageMapper.class),mock(SelfopOutboundOrderMapper.class),mock(SelfopReferenceRecovery.class),mock(SelfopValidationRunner.class),runs);
|
||||
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);
|
||||
}
|
||||
@Test void unexpectedValidationFailureTerminatesSessionInsteadOfSchedulingRetry() {
|
||||
executeValidationFailure(new IllegalStateException("VALIDATION_STRUCTURE_REJECTED"),"VALIDATION_STRUCTURE_REJECTED");
|
||||
}
|
||||
|
||||
+22
@@ -0,0 +1,22 @@
|
||||
package com.baiyue.lastleg.selfoperated.analytics.sync;
|
||||
|
||||
import com.baiyue.lastleg.selfoperated.analytics.mapper.*;
|
||||
import com.baiyue.lastleg.selfoperated.analytics.model.*;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import static com.baiyue.lastleg.selfoperated.analytics.model.SelfopSupport.utcNow;
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
import static org.mockito.Mockito.*;
|
||||
|
||||
class SelfopObservationReservationTest {
|
||||
@Test void consecutiveBatchReservationsRetainFenceAndSingleObservationSemantics() {
|
||||
var locks=mock(SelfopCoordinationMapper.class);var accounts=mock(SelfopOutboundDiscoveryScopeMapper.class);
|
||||
var row=new SelfopOutboundDiscoveryScope();row.setId(1L);row.setLeaseOwner("worker");row.setFenceToken(2L);
|
||||
row.setLeaseUntilUtc(utcNow().plusMinutes(1));row.setNextObservationSequence(101L);when(locks.lockDiscovery(1L)).thenReturn(row);
|
||||
var service=new SelfopLeaseService(locks,mock(SelfopSyncScopeMapper.class),accounts,new SelfopAnalyticsProperties());
|
||||
var lease=new SelfopLeaseService.Lease(1L,"worker",2L,true);
|
||||
assertEquals(101L,service.nextObservations(lease,100));assertEquals(201L,service.nextObservation(lease));assertEquals(202L,row.getNextObservationSequence());
|
||||
assertThrows(IllegalStateException.class,()->service.nextObservations(lease,0));assertThrows(IllegalStateException.class,()->service.nextObservations(lease,101));
|
||||
assertThrows(IllegalStateException.class,()->service.nextObservations(new SelfopLeaseService.Lease(1L,"old-worker",1L,true),10));
|
||||
assertEquals(202L,row.getNextObservationSequence());verify(accounts,times(2)).updateById(row);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user