diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundCandidateBuilder.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundCandidateBuilder.java index cb3fd7b08..d11a7035e 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundCandidateBuilder.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundCandidateBuilder.java @@ -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 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())) diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryService.java index 01d40237f..5a65e13d3 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryService.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryService.java @@ -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;index0){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> entries,int start) { + var candidates=new ArrayList>(); + for(int i=start;i(); + stages.selectList(new QueryWrapper().in("id",candidates.stream().map(e->e.getValue().getDirectoryStageId()).toList())) + .forEach(s->directory.put(s.getId(),s)); + var revised=new HashSet(); + stages.selectList(new QueryWrapper().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(); + 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().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); diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopLeaseService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopLeaseService.java index e019daf45..775e612a2 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopLeaseService.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopLeaseService.java @@ -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) { diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundCandidateBuilderTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundCandidateBuilderTest.java index d62d326a3..b1003ff90 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundCandidateBuilderTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundCandidateBuilderTest.java @@ -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)); diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryServiceTest.java index 18df3af68..52115e32c 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryServiceTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryServiceTest.java @@ -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{ + QueryWrapper 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.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()); diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopExecutionServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopExecutionServiceTest.java index 9a0914a9c..48f6ddcaa 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopExecutionServiceTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopExecutionServiceTest.java @@ -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"); } diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopObservationReservationTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopObservationReservationTest.java new file mode 100644 index 000000000..090860e78 --- /dev/null +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopObservationReservationTest.java @@ -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); + } +}