fix(selfop): publish complete single-pass inventory observations

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");
+                    require(manifest.getFirstScan().size()==manifest.getTotal(),"SELFOP_STOCK_COUNT_MISMATCH");
-                    if(manifest.getScan()==1){manifest.setScan(2);manifest.setNextPage(1);}else manifest.setDirectoryComplete(true);
+                    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);

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() {
+    @Test void missingRowCannotCompleteObservation() {
-        stableThreeRows(); config.setRequestBudget(3);
+        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)));
-        assertFalse(collector.collect(task, run, lease));
+        when(client.inventory(any(SelfopRequestContext.class), eq("A"), eq(2), eq(2))).thenReturn(page(2, 3, 2));
-        var cursor = read(task.getRoutingManifestJson(), SelfopRoutingManifest.class);
+        assertEquals("SELFOP_STOCK_COUNT_MISMATCH", assertThrows(IllegalStateException.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,
             () -> 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());

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() {