fix(selfop): restore unique same-second inventory chains
This commit is contained in:
+2
-2
@@ -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);
|
||||
|
||||
+64
-13
@@ -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<String> 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<BigDecimal>();
|
||||
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<Instant,BigDecimal>(Comparator.reverseOrder());
|
||||
@@ -114,14 +112,67 @@ public final class InventoryHistoryCalculator {
|
||||
}
|
||||
return null;
|
||||
}
|
||||
private static void paths(List<Flow> group,boolean[] used,BigDecimal balance,int depth,List<BigDecimal> endings) {
|
||||
if(endings.size()>1) return;
|
||||
if(depth==group.size()) { endings.add(balance); return; }
|
||||
for(int i=0;i<group.size();i++) if(!used[i]) {
|
||||
Flow f=group.get(i);
|
||||
if(balance!=null && balance.add(f.delta()).compareTo(f.closing())!=0) continue;
|
||||
used[i]=true;paths(group,used,f.closing(),depth+1,endings);used[i]=false;
|
||||
private record ChainResult(BigDecimal ending,String reason) { }
|
||||
private static ChainResult resolveChain(List<Flow> 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<BigDecimal,Integer>();
|
||||
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<Flow> group,BigDecimal start,boolean reverse) {
|
||||
var outgoing=new TreeMap<BigDecimal,ArrayDeque<Integer>>();
|
||||
for(int i=0;i<group.size();i++) {
|
||||
var f=group.get(i);
|
||||
outgoing.computeIfAbsent(f.closing().subtract(f.delta()),k->new ArrayDeque<>()).addLast(i);
|
||||
}
|
||||
var stack=new ArrayDeque<Integer>();
|
||||
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<Interval> intervals,Instant from,Instant until) {
|
||||
Instant cursor=from;
|
||||
|
||||
+90
@@ -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<Flow>();
|
||||
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<Flow>();
|
||||
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<Flow>();
|
||||
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<Integer> 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<Flow>();
|
||||
for(int i=0;i<edges.size();i++) {
|
||||
int edge=edges.get(i),opening=edge/3,closing=edge%3;
|
||||
group.add(flow("F"+i,GOOD,"PA",String.valueOf(closing-opening),String.valueOf(closing),"2026-10-15T01:00:00"));
|
||||
}
|
||||
for(int prior=-1;prior<3;prior++) {
|
||||
var rows=new ArrayList<>(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<Flow> group,boolean[] used,BigDecimal balance,int depth) {
|
||||
if(depth==group.size())return 1;
|
||||
int count=0;
|
||||
for(int i=0;i<group.size();i++)if(!used[i]) {
|
||||
var f=group.get(i);
|
||||
if(balance!=null&&balance.add(f.delta()).compareTo(f.closing())!=0)continue;
|
||||
used[i]=true;count+=countOrders(group,used,f.closing(),depth+1);used[i]=false;
|
||||
if(count>1)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"))));
|
||||
|
||||
+38
@@ -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<JsonNode>();
|
||||
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));
|
||||
|
||||
Reference in New Issue
Block a user