From 22d79e08f91dcadcacdf19a14b21b097462b260d Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Thu, 15 Oct 2026 14:13:41 +0800 Subject: [PATCH] fix(selfop): restore unique same-second inventory chains --- .../history/InventoryHistoryBuilder.java | 4 +- .../history/InventoryHistoryCalculator.java | 77 +++++++++++++--- .../InventoryHistoryCalculatorTest.java | 90 +++++++++++++++++++ .../history/InventoryHistoryPipelineTest.java | 38 ++++++++ 4 files changed, 194 insertions(+), 15 deletions(-) diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/history/InventoryHistoryBuilder.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/history/InventoryHistoryBuilder.java index f471266e0..1ffa3ad74 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/history/InventoryHistoryBuilder.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/history/InventoryHistoryBuilder.java @@ -109,12 +109,12 @@ public class InventoryHistoryBuilder { var sales=jdbc.queryForList("SELECT published_run_id FROM ll_selfop_sync_scope WHERE source_account=? AND warehouse_code=? AND data_family='OUTBOUND'",account,warehouse); Long salesRun=sales.isEmpty()||sales.get(0).get("published_run_id")==null?null:((Number)sales.get(0).get("published_run_id")).longValue(); String inputHash=versions.factHash(account,warehouse); - String signature=hash(Arrays.asList(inputHash,anchor,reference,from,until,salesRun,coverage)); + String signature=hash(Arrays.asList(RULE,CALCULATION_REVISION,inputHash,anchor,reference,from,until,salesRun,coverage)); if(jdbc.queryForObject("SELECT COUNT(*) FROM ll_selfop_sync_task WHERE source_account=? AND task_type=? AND request_hash=? AND status='SUCCEEDED'",Long.class,account,TYPE,signature)>0)return null; // Failed identical inputs get bounded attempts, instead of an infinite hot loop. if(jdbc.queryForObject("SELECT COUNT(*) FROM ll_selfop_sync_task WHERE source_account=? AND task_type=? AND request_hash=?",Long.class,account,TYPE,signature)>=3)return null; long taskId=IdWorker.getId(),runId=IdWorker.getId(),group=IdWorker.getId();var now=utcNow(); - String evidence=json(Map.of("warehouse",warehouse,"anchor",anchor,"reference",reference,"coverageRuns",coverageRuns,"inputStagesRun",runId,"rule",RULE)); + String evidence=json(Map.of("warehouse",warehouse,"anchor",anchor,"reference",reference,"coverageRuns",coverageRuns,"inputStagesRun",runId,"rule",RULE,"calculationRevision",CALCULATION_REVISION)); var task=new SelfopSyncTask();task.setId(taskId);task.setTaskNo("LXRESTORE-"+UUID.randomUUID().toString().replace("-","")); task.setSourceAccount(account);task.setDiscoveryScopeId(jdbc.queryForObject("SELECT id FROM ll_selfop_outbound_discovery_scope WHERE source_account=?",Long.class,account));task.setExecutionPurpose("BUSINESS"); task.setTaskRole("ACCOUNT_DISCOVERY");task.setTaskType(TYPE);task.setRequestKey("history-build:"+UUID.randomUUID());task.setRequestHash(signature); diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/history/InventoryHistoryCalculator.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/history/InventoryHistoryCalculator.java index ec4d54a15..746d164fa 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/history/InventoryHistoryCalculator.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/history/InventoryHistoryCalculator.java @@ -7,6 +7,7 @@ import java.util.*; /** Pure reconstruction over frozen, complete source intervals. Never invents absent stock. */ public final class InventoryHistoryCalculator { public static final String RULE = "FLOW_EOD_V1"; + public static final int CALCULATION_REVISION = 2; private static final Set VERIFIED_TYPES = Set.of("DO", "PA", "IMO", "IAT", "RI"); private static final ZoneOffset BUSINESS = ZoneOffset.ofHours(8); private InventoryHistoryCalculator() { } @@ -97,12 +98,9 @@ public final class InventoryHistoryCalculator { for(Flow f:chain) seconds.computeIfAbsent(f.at(),k->new ArrayList<>()).add(f); BigDecimal prior=null; for(var group:seconds.values()) { - if(group.size()>8) return "CHAIN_ORDER_AMBIGUOUS"; - var endings=new ArrayList(); - paths(group,new boolean[group.size()],prior,0,endings); - if(endings.isEmpty()) return "CHAIN_DISCONTINUITY"; - if(endings.size()!=1) return "CHAIN_ORDER_AMBIGUOUS"; - prior=endings.get(0); + var resolved=resolveChain(group,prior); + if(resolved.reason()!=null) return resolved.reason(); + prior=resolved.ending(); } } var deltas=new TreeMap(Comparator.reverseOrder()); @@ -114,14 +112,67 @@ public final class InventoryHistoryCalculator { } return null; } - private static void paths(List group,boolean[] used,BigDecimal balance,int depth,List endings) { - if(endings.size()>1) return; - if(depth==group.size()) { endings.add(balance); return; } - for(int i=0;i group,BigDecimal prior) { + // Every flow is a distinct directed edge: opening balance -> closing balance. + // A valid order is an Euler trail, using every edge exactly once. + var degree=new TreeMap(); + for(var f:group) { + degree.merge(f.closing().subtract(f.delta()),1,Integer::sum); + degree.merge(f.closing(),-1,Integer::sum); } + BigDecimal start=null;int starts=0,ends=0; + for(var entry:degree.entrySet()) { + int difference=entry.getValue(); + if(difference==1){start=entry.getKey();starts++;} + else if(difference==-1)ends++; + else if(difference!=0)return new ChainResult(null,"CHAIN_DISCONTINUITY"); + } + if(!((starts==1&&ends==1)||(starts==0&&ends==0)))return new ChainResult(null,"CHAIN_DISCONTINUITY"); + if(prior!=null) { + if(start!=null&&start.compareTo(prior)!=0)return new ChainResult(null,"CHAIN_DISCONTINUITY"); + start=prior; + } else if(start==null)start=group.get(0).closing().subtract(group.get(0).delta()); + var first=orderedTrail(group,start,false); + if(first==null)return new ChainResult(null,"CHAIN_DISCONTINUITY"); + // Without a prior balance a circuit with multiple edges admits rotated orders. + if(prior==null&&starts==0&&group.size()>1)return new ChainResult(null,"CHAIN_ORDER_AMBIGUOUS"); + // Hierholzer with opposite edge priorities yields the lexicographically first + // and last complete trails. Equality proves uniqueness, including parallel edges. + // Input indices only define opposing priorities; neither IDs nor response order + // are accepted as chronology. Two iterative passes avoid factorial search/recursion. + var last=orderedTrail(group,start,true); + if(!Arrays.equals(first,last))return new ChainResult(null,"CHAIN_ORDER_AMBIGUOUS"); + return new ChainResult(group.get(first[first.length-1]).closing(),null); + } + private static int[] orderedTrail(List group,BigDecimal start,boolean reverse) { + var outgoing=new TreeMap>(); + for(int i=0;inew ArrayDeque<>()).addLast(i); + } + var stack=new ArrayDeque(); + int[] trail=new int[group.size()];int position=trail.length; + BigDecimal balance=start; + while(true) { + var edges=outgoing.get(balance); + if(edges!=null&&!edges.isEmpty()) { + int edge=reverse?edges.removeLast():edges.removeFirst(); + stack.push(edge);balance=group.get(edge).closing(); + } else { + if(stack.isEmpty())break; + int edge=stack.pop();trail[--position]=edge; + var f=group.get(edge);balance=f.closing().subtract(f.delta()); + } + } + if(position!=0)return null; // Disconnected edges cannot be silently discarded. + balance=start; + for(int edge:trail) { + var f=group.get(edge); + if(balance.add(f.delta()).compareTo(f.closing())!=0)return null; + balance=f.closing(); + } + return trail; } public static boolean covered(List intervals,Instant from,Instant until) { Instant cursor=from; diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/history/InventoryHistoryCalculatorTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/history/InventoryHistoryCalculatorTest.java index c3921f5d4..4316a78e4 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/history/InventoryHistoryCalculatorTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/history/InventoryHistoryCalculatorTest.java @@ -69,6 +69,96 @@ class InventoryHistoryCalculatorTest { var rows=List.of(flow("1",GOOD,"DO","-1","38","2026-10-15T01:00:00"),flow("2",GOOD,"DO","-1","35","2026-10-15T02:00:00")); assertThat(reconstruct(input(rows,List.of(anchor(GOOD,"35")))).get(0).reason()).isEqualTo("CHAIN_DISCONTINUITY"); } + @Test void restoresHundredSameSecondDecrementsAcrossSixtyDaysRegardlessOfSourceOrder() { + var group=new ArrayList(); + for(int i=0;i<100;i++)group.add(flow("F"+(100-i),GOOD,"DO","-1",String.valueOf(726-i),"2026-10-15T01:00:00")); + for(int seed=0;seed<10;seed++) { + Collections.shuffle(group,new Random(seed)); + for(boolean hasPrior:List.of(false,true)) { + var rows=new ArrayList<>(group); + if(hasPrior)rows.add(flow("prior",GOOD,"DO","-1","727","2026-10-14T23:59:59")); + var i=input(rows,List.of(anchor(GOOD,"627"))); + var result=reconstruct(new Input(DAY.minusDays(59),i.until(),i.candidates(),i.flows(),i.anchors(), + List.of(new Interval(DAY.minusDays(59).atStartOfDay().toInstant(ZoneOffset.ofHours(8)),at("2026-10-15T08:00:02"))))); + assertThat(result).hasSize(60).allMatch(d->"VALID".equals(d.status())); + assertThat(result.get(59).quantity()).isEqualByComparingTo("727"); + assertThat(result.get(0).quantity()).isEqualByComparingTo(hasPrior?"728":"727"); + } + } + } + @Test void forcedCycleBeforeExitIsUniqueDespiteMultipleOutgoingEdgesAndDecimalScales() { + // Taking the exit first is a dead end, not a second complete order. + var rows=List.of(flow("exit",GOOD,"DO","-1.00","0.0","2026-10-15T01:00:00"), + flow("up",GOOD,"PA","1","2.000","2026-10-15T01:00:00"), + flow("down",GOOD,"DO","-1","1.000000","2026-10-15T01:00:00")); + assertThat(reconstruct(input(rows,List.of(anchor(GOOD,"10")))).get(0).status()).isEqualTo("VALID"); + } + @Test void twoCyclesWithSameFinalBalanceRemainAmbiguous() { + var rows=List.of(flow("prior",GOOD,"PA","0","2","2026-10-15T00:59:59"), + flow("a",GOOD,"PA","1","3","2026-10-15T01:00:00"),flow("b",GOOD,"DO","-1","2","2026-10-15T01:00:00"), + flow("c",GOOD,"PA","2","4","2026-10-15T01:00:00"),flow("d",GOOD,"DO","-2","2","2026-10-15T01:00:00")); + assertThat(reconstruct(input(rows,List.of(anchor(GOOD,"10")))).get(0).reason()).isEqualTo("CHAIN_ORDER_AMBIGUOUS"); + } + @Test void distinctIdsWithDuplicateEdgesAreNotDeduplicatedIntoAUniqueOrder() { + var rows=List.of(flow("a",GOOD,"PA","1","2","2026-10-15T01:00:00"), + flow("b",GOOD,"PA","1","2","2026-10-15T01:00:00"),flow("return",GOOD,"DO","-1","1","2026-10-15T01:00:00")); + assertThat(reconstruct(input(rows,List.of(anchor(GOOD,"10")))).get(0).reason()).isEqualTo("CHAIN_ORDER_AMBIGUOUS"); + } + @Test void disconnectedCycleCannotBeIgnoredEvenWhenDegreesMatch() { + var rows=List.of(flow("path",GOOD,"DO","-1","1","2026-10-15T01:00:00"), + flow("loop",GOOD,"PA","0","3","2026-10-15T01:00:00")); + assertThat(reconstruct(input(rows,List.of(anchor(GOOD,"10")))).get(0).reason()).isEqualTo("CHAIN_DISCONTINUITY"); + } + @Test void longChainWithMissingEdgeOrWrongPriorRemainsUnknown() { + var rows=new ArrayList(); + for(int i=0;i<100;i++)rows.add(flow("F"+i,GOOD,"DO","-1",String.valueOf(726-i),"2026-10-15T01:00:00")); + rows.add(flow("prior",GOOD,"PA","0","728","2026-10-15T00:59:59")); + assertThat(reconstruct(input(rows,List.of(anchor(GOOD,"627")))).get(0).reason()).isEqualTo("CHAIN_DISCONTINUITY"); + rows.remove(rows.size()-1);rows.remove(50); + assertThat(reconstruct(input(rows,List.of(anchor(GOOD,"627")))).get(0).reason()).isEqualTo("CHAIN_DISCONTINUITY"); + } + @Test void largeUniqueChainUsesIterativeTraversal() { + var rows=new ArrayList(); + for(int i=0;i<20_000;i++)rows.add(flow("F"+i,GOOD,"DO","-1",String.valueOf(20_000-i),"2026-10-15T01:00:00")); + assertThat(reconstruct(input(rows,List.of(anchor(GOOD,"1")))).get(0).quantity()).isEqualByComparingTo("20001"); + } + @Test void graphSolverMatchesExhaustiveOrdersForAllSmallMultigraphs() { + // Independent permutation oracle: all multisets of up to five directed edges + // on three balances, including loops/parallel edges, with every possible prior. + for(int size=1;size<=5;size++)checkGraphs(new ArrayList<>(),0,size); + } + private void checkGraphs(List edges,int minimum,int remaining) { + if(remaining>0) { + for(int edge=minimum;edge<9;edge++) { + edges.add(edge);checkGraphs(edges,edge,remaining-1);edges.remove(edges.size()-1); + } + return; + } + var group=new ArrayList(); + for(int i=0;i(group); + if(prior>=0)rows.add(flow("prior",GOOD,"PA","0",String.valueOf(prior),"2026-10-15T00:59:59")); + int paths=countOrders(group,new boolean[group.size()],prior<0?null:BigDecimal.valueOf(prior),0); + String reason=paths==0?"CHAIN_DISCONTINUITY":paths==1?null:"CHAIN_ORDER_AMBIGUOUS"; + var actual=reconstruct(input(rows,List.of(anchor(GOOD,"100")))).get(0); + assertThat(actual.reason()).as("edges=%s prior=%s",edges,prior).isEqualTo(reason); + } + } + private int countOrders(List group,boolean[] used,BigDecimal balance,int depth) { + if(depth==group.size())return 1; + int count=0; + for(int i=0;i1)return 2; + } + return count; + } @Test void laterAnchorRevisionChangesEarlierDay() { var a=reconstruct(input(List.of(flow("f",GOOD,"DO","-1","100","2026-10-15T01:00:00")),List.of(anchor(GOOD,"100")))); var b=reconstruct(input(List.of(flow("f",GOOD,"DO","-2","100","2026-10-15T01:00:00")),List.of(anchor(GOOD,"100")))); diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/history/InventoryHistoryPipelineTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/history/InventoryHistoryPipelineTest.java index 7c6b77e7b..48d78a109 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/history/InventoryHistoryPipelineTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/history/InventoryHistoryPipelineTest.java @@ -118,6 +118,44 @@ public class InventoryHistoryPipelineTest { assertEquals(2,jdbc.queryForObject("SELECT COUNT(*) FROM ll_selfop_inventory_snapshot",Integer.class)); builder.build("demo","W",30);assertEquals(1,jdbc.queryForObject("SELECT COUNT(*) FROM ll_selfop_inventory_day_version",Integer.class)); } + @Test void calculationRevisionRebuildsLegacyUnknownFromSavedFlowsExactlyOnce() { + stock(100,DAY,727);stock(101,DAY.plusDays(1),627); + var rows=new ArrayList(); + for(int i=0;i<100;i++)rows.add(flow("F"+i,"-1",String.valueOf(726-i))); + Collections.shuffle(rows,new Random(42)); + collect(slice(),DAY,page(1,100,rows)); + collect(new InventoryFlowSlice("W",DAY.plusDays(1).atStartOfDay(),DAY.plusDays(1).atTime(8,0,2),1000),DAY,page(1,0,List.of())); + builder.build("demo","W",30); + // Seed the pre-fix published state, including its exact former dedup signature. + var coverage=List.of(new InventoryHistoryCalculator.Interval( + DAY.atStartOfDay().toInstant(ZoneOffset.ofHours(8)),DAY.plusDays(1).atTime(8,0,2).toInstant(ZoneOffset.ofHours(8)))); + String legacyHash=hash(Arrays.asList(versions.factHash("demo","W"),builder.anchor("demo","W",0), + builder.anchor("demo","W",1),DAY,DAY.plusDays(1),null,coverage)); + var legacyTask=jdbc.queryForMap("SELECT id,request_hash FROM ll_selfop_sync_task WHERE task_type='INVENTORY_HISTORY_BUILD'"); + assertNotEquals(legacyHash,legacyTask.get("request_hash")); + jdbc.update("UPDATE ll_selfop_sync_task SET request_hash=? WHERE id=?",legacyHash,legacyTask.get("id")); + jdbc.update("UPDATE ll_selfop_inventory_daily SET physical_quantity=NULL,quality_status='UNKNOWN',quality_reason='CHAIN_ORDER_AMBIGUOUS'"); + jdbc.update("UPDATE ll_selfop_inventory_day_version SET evidence_json=?",json(Map.of("rule","FLOW_EOD_V1"))); + var flowHash=versions.factHash("demo","W"); + clearInvocations(client); + builder.build("demo","W",30); + assertEquals(2,jdbc.queryForObject("SELECT COUNT(*) FROM ll_selfop_sync_task WHERE task_type='INVENTORY_HISTORY_BUILD' AND status='SUCCEEDED'",Integer.class)); + assertEquals(1,jdbc.queryForObject("SELECT COUNT(*) FROM ll_selfop_inventory_day_version WHERE head_slot=1",Integer.class)); + var current=jdbc.queryForMap("SELECT d.physical_quantity,d.quality_status,v.rule_version,v.evidence_json FROM ll_selfop_inventory_daily d JOIN ll_selfop_inventory_day_version v ON v.id=d.day_version_id WHERE v.head_slot=1"); + assertEquals(0,new BigDecimal("627").compareTo((BigDecimal)current.get("physical_quantity"))); + assertEquals("VALID",current.get("quality_status")); + assertEquals("FLOW_EOD_V1",current.get("rule_version")); + var evidence=read(current.get("evidence_json").toString(),JsonNode.class); + assertEquals(InventoryHistoryCalculator.CALCULATION_REVISION,evidence.path("calculationRevision").asInt()); + var runEvidence=read(jdbc.queryForObject("SELECT scope_snapshot_json FROM ll_selfop_sync_run WHERE status='FACTS_APPLIED' AND last_phase='HISTORY_PUBLISHED' ORDER BY id DESC LIMIT 1",String.class),JsonNode.class); + assertEquals(InventoryHistoryCalculator.CALCULATION_REVISION,runEvidence.path("calculationRevision").asInt()); + builder.build("demo","W",30); + assertEquals(2,jdbc.queryForObject("SELECT COUNT(*) FROM ll_selfop_inventory_day_version",Integer.class)); + assertEquals(flowHash,versions.factHash("demo","W")); + assertEquals(100,jdbc.queryForObject("SELECT COUNT(*) FROM ll_selfop_inventory_flow",Integer.class)); + assertEquals(2,jdbc.queryForObject("SELECT COUNT(*) FROM ll_selfop_inventory_snapshot",Integer.class)); + verifyNoInteractions(client); + } void stock(long id,LocalDate day,int quantity) { var now=day.atTime(0,0);var r=new SelfopSyncRun();r.setId(id);r.setTaskId(id);r.setAttemptNo(1);r.setSourceAccount("demo");r.setWarehouseCode("W");r.setScopeId(2L);r.setRunRole("WAREHOUSE_SYNC");r.setExecutionPurpose("BUSINESS");r.setScopeHash("b".repeat(64));r.setScopeSnapshotJson("{}");r.setFenceToken(0L);r.setStatus("PUBLISHED");r.setLastPhase("PUBLISHED");r.setCoverageStatus("COMPLETE");r.setSourceTotalCount(1L);r.setUniqueObjectCount(1L);r.setPageCount(1);r.setCreatedAt(now);r.setUpdatedAt(now);r.setFetchStartedAtUtc(now);r.setFetchFinishedAtUtc(now.plusSeconds(1));r.setPublishedAtUtc(now.plusSeconds(2));runs.insert(r); jdbc.update("INSERT INTO ll_selfop_inventory_snapshot(id,created_at,updated_at,snapshot_run_id,source_account,customer_code,warehouse_code,sku_code,product_quality,physical_quantity,observed_at_utc,quality_status) VALUES(?,?,?,?,'demo','C','W','SKU',0,?,?,'VALID')",id,now,now,id,quantity,now.plusSeconds(1));