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 e8a8441c1..67be22b79 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 @@ -1,6 +1,5 @@ package com.baiyue.lastleg.selfoperated.analytics.sync; -import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; import com.baomidou.mybatisplus.core.toolkit.IdWorker; import com.baiyue.lastleg.selfoperated.analytics.client.LingxingAnalyticsClient; import com.baiyue.lastleg.selfoperated.analytics.mapper.*; @@ -40,17 +39,15 @@ public class SelfopInventoryCollector { require(task.getSourceAccount().equals(row.sourceAccount())&&task.getWarehouseCode().equals(row.warehouseCode()),"SELFOP_STOCK_SCOPE_LEAK"); identity(row.customerCode(),64);identity(row.skuCode(),128);require(row.productQuality()!=null,"SELFOP_STOCK_IDENTITY_MISSING"); String key=hash(List.of(row.sourceAccount(),row.customerCode(),row.warehouseCode(),row.skuCode(),row.productQuality())); - if(manifest.getScan()==1){require(!manifest.getFirstScan().containsKey(key),"SELFOP_STOCK_DUPLICATE");manifest.getFirstScan().put(key,SelfopStockFingerprint.of(row)); + require(!manifest.getFirstScan().containsKey(key),"SELFOP_STOCK_DUPLICATE");manifest.getFirstScan().put(key,SelfopStockFingerprint.of(row)); var stage=new SelfopSyncStage();stage.setId(IdWorker.getId());stage.setRunId(run.getId());stage.setStageBatchId(run.getId());stage.setLastFetchRunId(run.getId()); stage.setWarehouseCode(row.warehouseCode());stage.setObjectType("PRODUCT_STOCK");stage.setObjectKeyHash(key);stage.setFirstPageNo(manifest.getNextPage()); stage.setObservedAtUtc(utcNow());stage.setContentHash(SelfopStockFingerprint.of(row));stage.setPayloadJson(json(row));stage.setValidationStatus("VALID");stage.setCreatedAt(utcNow());stage.setUpdatedAt(utcNow());stages.insert(stage); - }else{require(Objects.equals(manifest.getFirstScan().get(key),SelfopStockFingerprint.of(row)),"SELFOP_STOCK_SCAN_UNSTABLE"); - require(!manifest.getObjects().containsKey(key),"SELFOP_STOCK_DUPLICATE");manifest.getObjects().put(key,new SelfopRoutingManifest.Entry());} } manifest.setNextPage(manifest.getNextPage()+1); if(manifest.getPages()==0||manifest.getNextPage()>manifest.getPages()){ - require((manifest.getScan()==1?manifest.getFirstScan().size():manifest.getObjects().size())==manifest.getTotal(),"SELFOP_STOCK_COUNT_MISMATCH"); - if(manifest.getScan()==1){manifest.setScan(2);manifest.setNextPage(1);}else manifest.setDirectoryComplete(true); + require(manifest.getFirstScan().size()==manifest.getTotal(),"SELFOP_STOCK_COUNT_MISMATCH"); + manifest.setDirectoryComplete(true); } task.setRoutingManifestJson(json(manifest));task.setUpdatedAt(utcNow());tasks.updateById(task); SelfopProgressProjection.applyCheckpoint(task,run,"INVENTORY","PAGE",Math.max(0,manifest.getNextPage()-1),manifest.getPages()==null?null:manifest.getPages().longValue(),manifest.getScan());tasks.updateById(task);runs.updateById(run); 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 a4716df43..94a32f4fd 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 @@ -54,18 +54,18 @@ class SelfopInventoryCollectorTest { when(client.inventory(any(SelfopRequestContext.class), eq("A"), eq(2), eq(2))).thenReturn(page(2, 3, 2, stock("c", null))); } - @Test void twoIdenticalScansValidateOneAttemptAndOnlyFirstScanWritesEvidence() { + @Test void oneCompleteObservationValidatesAttemptAndPersistsEachPageOnce() { stableThreeRows(); assertTrue(collector.collect(task, run, lease)); assertEquals("VALIDATED", run.getStatus()); assertEquals("COMPLETE", run.getCoverageStatus()); assertEquals(3L, run.getSourceTotalCount()); assertEquals(3L, run.getUniqueObjectCount()); assertEquals(2, run.getPageCount()); - verify(client, times(2)).inventory(any(SelfopRequestContext.class), eq("A"), eq(1), eq(2)); verify(client, times(2)).inventory(any(SelfopRequestContext.class), eq("A"), eq(2), eq(2)); + verify(client).inventory(any(SelfopRequestContext.class), eq("A"), eq(1), eq(2)); verify(client).inventory(any(SelfopRequestContext.class), eq("A"), eq(2), eq(2)); var capture = ArgumentCaptor.forClass(SelfopSyncStage.class); verify(stages, times(3)).insert(capture.capture()); assertTrue(capture.getAllValues().stream().allMatch(s -> s.getRunId() == 10L && s.getStageBatchId() == 10L && s.getLastFetchRunId() == 10L)); assertNotNull(run.getFetchFinishedAtUtc()); var contexts=ArgumentCaptor.forClass(SelfopRequestContext.class); - verify(client,times(4)).inventory(contexts.capture(),eq("A"),anyInt(),eq(2)); + verify(client,times(2)).inventory(contexts.capture(),eq("A"),anyInt(),eq(2)); assertTrue(contexts.getAllValues().stream().allMatch(c->"demo".equals(c.sourceAccount())&&"BUSINESS".equals(c.executionPurpose()) &&Objects.equals(c.taskId(),20L)&&Objects.equals(c.runId(),10L)&&Objects.equals(c.rootTaskId(),20L))); } @@ -112,28 +112,31 @@ class SelfopInventoryCollectorTest { assertTrue(restarted.collect(task, run, lease)); var capture = ArgumentCaptor.forClass(SelfopSyncStage.class); verify(stages, times(3)).insert(capture.capture()); assertTrue(capture.getAllValues().stream().allMatch(s -> Objects.equals(s.getStageBatchId(), 10L))); - verify(client, times(2)).inventory(any(SelfopRequestContext.class), eq("A"), eq(1), eq(2)); verify(client, times(2)).inventory(any(SelfopRequestContext.class), eq("A"), eq(2), eq(2)); + verify(client).inventory(any(SelfopRequestContext.class), eq("A"), eq(1), eq(2)); verify(client).inventory(any(SelfopRequestContext.class), eq("A"), eq(2), eq(2)); assertEquals("COMPLETE", run.getCoverageStatus()); } - @Test void secondScanCanYieldAndResumeWithoutDuplicatingFirstScanStages() { - stableThreeRows(); config.setRequestBudget(3); - assertFalse(collector.collect(task, run, lease)); - var cursor = read(task.getRoutingManifestJson(), SelfopRoutingManifest.class); - assertEquals(2, cursor.getScan()); assertEquals(2, cursor.getNextPage()); assertEquals(2, cursor.getObjects().size()); - config.setRequestBudget(10); assertTrue(collector.collect(task, run, lease)); - verify(stages, times(3)).insert(any(SelfopSyncStage.class)); - verify(client, times(2)).inventory(any(SelfopRequestContext.class), eq("A"), eq(1), eq(2)); verify(client, times(2)).inventory(any(SelfopRequestContext.class), eq("A"), eq(2), eq(2)); - } - @Test void secondScanChangedQuantityCannotPassByKeepingSameIdentitySet() { - when(client.inventory(any(SelfopRequestContext.class), eq("A"), eq(1), eq(2))).thenReturn(page(1, 1, 1, stock("a", BigDecimal.ONE)), - page(1, 1, 1, stock("a", BigDecimal.TEN))); - assertEquals("SELFOP_STOCK_SCAN_UNSTABLE", assertThrows(IllegalStateException.class, + @Test void missingRowCannotCompleteObservation() { + 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)); + assertEquals("SELFOP_STOCK_COUNT_MISMATCH", assertThrows(IllegalStateException.class, () -> collector.collect(task, run, lease)).getMessage()); assertIncompleteAttemptWithPersistedProgress(); } - @Test void equivalentScientificAndDecimalQuantitiesPassBothScans() { + @Test void completeObservationDoesNotReReadChangingSource() { + var started = run.getFetchStartedAtUtc(); + when(client.inventory(any(SelfopRequestContext.class), eq("A"), eq(1), eq(2))).thenReturn(page(1, 1, 1, stock("a", BigDecimal.ONE)), + page(1, 1, 1, stock("a", BigDecimal.TEN))); + assertTrue(collector.collect(task, run, lease)); + verify(client).inventory(any(SelfopRequestContext.class), eq("A"), eq(1), eq(2)); + var captured = ArgumentCaptor.forClass(SelfopSyncStage.class); verify(stages).insert(captured.capture()); + assertEquals(BigDecimal.ONE, read(captured.getValue().getPayloadJson(), SelfopPayloads.Stock.class).availableQuantity()); + assertEquals(started, run.getFetchStartedAtUtc()); + assertFalse(run.getFetchFinishedAtUtc().isBefore(started)); + assertTrue(read(task.getRoutingManifestJson(), SelfopRoutingManifest.class).isDirectoryComplete()); + } + @Test void scientificQuantityPersistsCanonicalFingerprint() { when(client.inventory(any(SelfopRequestContext.class),eq("A"),eq(1),eq(2))).thenReturn( - page(1,1,1,stock("a",new BigDecimal("1E+2"))),page(1,1,1,stock("a",new BigDecimal("100.0")))); + page(1,1,1,stock("a",new BigDecimal("1E+2")))); assertTrue(collector.collect(task,run,lease)); assertEquals("COMPLETE",run.getCoverageStatus()); var captured=ArgumentCaptor.forClass(SelfopSyncStage.class);verify(stages).insert(captured.capture()); 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 e5ecda812..576bc09b3 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 @@ -94,7 +94,7 @@ class SelfopInventoryRecoveryTest { assertTrue(collector.collect(task, fresh, lease)); var saved = ArgumentCaptor.forClass(SelfopSyncStage.class); verify(stages).insert(saved.capture()); assertEquals(fresh.getId(), saved.getValue().getStageBatchId()); assertEquals(fresh.getId(), saved.getValue().getRunId()); - assertEquals(1, saved.getValue().getFirstPageNo()); verify(client, times(2)).inventory(any(SelfopRequestContext.class), eq("A"), eq(1), eq(2)); + assertEquals(1, saved.getValue().getFirstPageNo()); verify(client).inventory(any(SelfopRequestContext.class), eq("A"), eq(1), eq(2)); } @Test void exhaustedFailuresRetainOldPublishedPointerAndCannotCreateAnotherAttempt() {