From 913b9e92629c2f4951e5fcd712f9e6953c02394b Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Wed, 14 Oct 2026 04:59:05 +0800 Subject: [PATCH] fix(selfop): persist bounded next-root quantity diagnostics --- .../LingxingQuantityCaptureProperties.java | 17 +- .../lingxing/LingxingQuantityCapture.java | 259 ++++----- .../lingxing/LingxingQuantityEvidence.java | 41 +- .../lingxing/LingxingQuantityHttpSink.java | 83 ++- .../provider/lingxing/LingxingWmsClient.java | 4 +- .../lingxing/LingxingQuantityCaptureTest.java | 180 ++---- .../LingxingQuantityHttpSinkTest.java | 85 +-- .../lingxing/LingxingWmsClientTest.java | 7 +- .../QuantityDiagnosticContract.java | 26 +- .../QuantityDiagnosticController.java | 41 +- .../diagnostic/QuantityDiagnosticPolicy.java | 106 +--- .../QuantityDiagnosticProperties.java | 7 +- .../diagnostic/QuantityDiagnosticService.java | 284 ++++++--- .../db/selfop-quantity-diagnostics.sql | 27 +- ...lfop-quantity-diagnostics-internal-v1.yaml | 549 ++++++++++++++++-- .../QuantityDiagnosticControllerTest.java | 58 +- .../QuantityDiagnosticServiceTest.java | 282 ++++----- 17 files changed, 1142 insertions(+), 914 deletions(-) diff --git a/integration-service/src/main/java/com/baiyue/integration/config/LingxingQuantityCaptureProperties.java b/integration-service/src/main/java/com/baiyue/integration/config/LingxingQuantityCaptureProperties.java index 1e94414c2..95d4099e1 100644 --- a/integration-service/src/main/java/com/baiyue/integration/config/LingxingQuantityCaptureProperties.java +++ b/integration-service/src/main/java/com/baiyue/integration/config/LingxingQuantityCaptureProperties.java @@ -1,19 +1,6 @@ package com.baiyue.integration.config; - -import java.time.Instant; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.boot.context.properties.bind.DefaultValue; - -/** Fixed one-shot scope. No refresh/restart may extend the configured absolute window. */ +/** Feature gate only. Explicit, durable grants and deadlines are owned by Lastleg. */ @ConfigurationProperties("baiyue.integration.lingxing-wms.quantity-capture") -public record LingxingQuantityCaptureProperties(@DefaultValue("false") boolean enabled, - String captureId, String sourceAccount, String rootTaskId, String sessionId, - String warehouseCode, String customerCode, Instant startsAt, Instant expiresAt) { - public boolean valid() { - return text(captureId) && text(sourceAccount) && text(rootTaskId) && text(sessionId) - && text(warehouseCode) && text(customerCode) && startsAt != null && expiresAt != null - && expiresAt.isAfter(startsAt) && !expiresAt.isAfter(startsAt.plusSeconds(1800)); - } - private static boolean text(String value) { return value != null && !value.isBlank() && value.length() <= 128; } - @Override public String toString() { return "LingxingQuantityCaptureProperties[enabled=" + enabled + "]"; } -} +public record LingxingQuantityCaptureProperties(@DefaultValue("false") boolean enabled,String sourceAccount) {} diff --git a/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingQuantityCapture.java b/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingQuantityCapture.java index 8500b810c..976cc06df 100644 --- a/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingQuantityCapture.java +++ b/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingQuantityCapture.java @@ -1,171 +1,122 @@ package com.baiyue.integration.provider.lingxing; +import static com.baiyue.integration.provider.lingxing.LingxingQuantityEvidence.*; import com.baiyue.integration.config.LingxingQuantityCaptureProperties; import com.baiyue.integration.dto.lingxing.LingxingEgressDtos.EgressContext; +import com.baiyue.integration.dto.lingxing.LingxingEgressDtos.Operation; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import jakarta.annotation.PreDestroy; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import org.springframework.beans.factory.ObjectProvider; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import java.math.BigDecimal; -import java.time.Clock; -import java.time.Instant; +import java.time.*; import java.util.*; import java.util.concurrent.*; -import static com.baiyue.integration.provider.lingxing.LingxingQuantityEvidence.*; +import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.ReentrantLock; -/** Passive bounded observer. It has no supplier client and never changes collection decisions. */ +/** Passive whitelist snapshots only. No JSON serialization, HTTP or waiting locks on egress threads. */ @Component public final class LingxingQuantityCapture { - private static final Logger LOG = LoggerFactory.getLogger(LingxingQuantityCapture.class); - private static final int MAX_ORDERS = 100, MAX_ROWS = 2000, MAX_BYTES = 2 * 1024 * 1024; - private final LingxingQuantityCaptureProperties config; - private final ObjectMapper mapper; - private final Sink sink; - private final Clock clock; - private final ThreadPoolExecutor worker; - private final Map pending = new HashMap<>(); - private final Set completed = new HashSet<>(); - private volatile String stopReason; - private int records, rows, bytes; - - @Autowired - public LingxingQuantityCapture(LingxingQuantityCaptureProperties config, ObjectMapper mapper, - ObjectProvider sink) { - this(config, mapper, sink.getIfAvailable(), Clock.systemUTC()); - } - LingxingQuantityCapture(LingxingQuantityCaptureProperties config, ObjectMapper mapper, Sink sink, Clock clock) { - this.config = config; this.mapper = mapper; this.sink = sink; this.clock = clock; - worker = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS, new ArrayBlockingQueue<>(MAX_ORDERS), task -> { - Thread thread = new Thread(task, "lingxing-quantity-capture"); thread.setDaemon(true); return thread; - }, new ThreadPoolExecutor.AbortPolicy()); - if (!config.enabled()) stopReason = "DISABLED"; - else if (!config.valid() || sink == null) stopReason = "CONFIGURATION_UNAVAILABLE"; - } - - /** Called only after the normal projection has successfully validated a supplier response. */ - public synchronized void observe(JsonNode data, boolean detail, EgressContext context, Instant observedAt) { - try { - if (!active() || context == null || !context.wellFormed() - || !Objects.equals(config.sourceAccount(), context.sourceAccount()) - || !Objects.equals(config.rootTaskId(), context.rootTaskId()) - || !Objects.equals(config.sessionId(), context.sessionId())) return; - if (detail) enqueue(data, true, context, observedAt); - else for (JsonNode row : data.path("records")) { - if (!active()) break; - enqueue(row, false, context, observedAt); - } - } catch (RuntimeException error) { stop("OBSERVATION_FAILED"); } - } - private void enqueue(JsonNode data, boolean detail, EgressContext context, Instant time) { - String warehouse = text(data, "whCode", 64), customer = text(data, "customerCode", 64); - if (!("*".equals(config.warehouseCode()) || config.warehouseCode().equals(warehouse)) - || !("*".equals(config.customerCode()) || config.customerCode().equals(customer))) return; - Key key = new Key(warehouse, customer, text(data, "deliveryNo", 128)); - if (completed.contains(key)) return; - // Freeze the first 100 list candidates. Extra page rows and unselected details cost no retained memory. - if (detail && !pending.containsKey(key)) return; - if (!detail && (pending.containsKey(key) || pending.size() + completed.size() >= MAX_ORDERS)) return; - JsonNode items = data.path(detail ? "skuVOList" : "productList"); - if (!items.isArray() || items.isEmpty() || items.size() > 200) { stop("SKU_RECORD_LIMIT_OR_MISSING"); return; } - Map quantities = new TreeMap<>(); - for (JsonNode item : items) { - String sku = text(item, "productSku", 128); - Quantities next = new Quantities(detail ? null : quantity(item.get("qty")), - detail ? quantity(item.get("deliveryQty")) : null, - detail ? quantity(item.get("actualDeliveryQty")) : null); - quantities.merge(sku, next, Quantities::add); - } - Observation observation = new Observation(data.path("status").isInt() ? data.path("status").asText() : null, - optionalText(data, "outboundTime", 64), time, context.attemptId(), true); - Sample sample = new Sample(key, detail, observation, quantities, items.size()); - process(sample); - } - private void process(Sample sample) { - try { - if (!active()) return; - if (++records > 2000 || (rows += sample.rowCount()) > MAX_ROWS - || (bytes += mapper.writeValueAsBytes(sample).length) > MAX_BYTES) { stop("CAPTURE_LIMIT"); return; } - if (completed.contains(sample.key())) return; - // Directory and detail stages can be far apart: candidates live until the fixed session expiry. - Pending pair = pending.computeIfAbsent(sample.key(), ignored -> new Pending()); - // Keep the first observation on each side; retries cannot silently replace its generation. - if (sample.detail()) { if (pair.detail == null) pair.detail = sample; } - else if (pair.list == null) pair.list = sample; - if (pair.list == null || pair.detail == null) return; - Set skus = new TreeSet<>(pair.list.quantities().keySet()); skus.addAll(pair.detail.quantities().keySet()); - if (skus.size() > 200) { stop("ORDER_SKU_LIMIT"); return; } - List lines = new ArrayList<>(); boolean difference = false; - for (String sku : skus) { - Quantities list = pair.list.quantities().get(sku), detail = pair.detail.quantities().get(sku); - BigDecimal forecast = detail == null ? null : detail.delivery, actual = detail == null ? null : detail.actual; - difference |= forecast != null && actual != null && forecast.compareTo(actual) != 0; - lines.add(new Line(sku, list != null, detail != null, list == null ? null : list.list, forecast, actual)); - } - Key key = sample.key(); - LingxingQuantityEvidence evidence = new LingxingQuantityEvidence(config.captureId(), config.sourceAccount(), - config.rootTaskId(), config.sessionId(), key.warehouse(), key.customer(), key.order(), - pair.list.observation(), pair.detail.observation(), List.copyOf(lines)); - worker.execute(() -> { - try { - if (!clock.instant().isBefore(config.expiresAt())) { abortTransmission("EXPIRED"); return; } - sink.accept(evidence); - } catch (ReceiverStopped normalStop) { abortTransmission(normalStop.reason()); } - catch (Exception failure) { abortTransmission("DIAGNOSTIC_FAILED"); } - }); - completed.add(key); pending.remove(key); - if (difference) stop("FORECAST_ACTUAL_DIFFERENCE"); - else if (completed.size() >= MAX_ORDERS) stop("ORDER_LIMIT"); - } catch (Exception error) { stop("DIAGNOSTIC_FAILED"); } - } - private boolean active() { - if (stopReason != null) return false; - if (!clock.instant().isBefore(config.expiresAt())) { stop("EXPIRED"); return false; } - return !clock.instant().isBefore(config.startsAt()); - } - private synchronized void stop(String reason) { - if (stopReason != null) return; - stopReason = reason; - if (!"ORDER_LIMIT".equals(reason) && !"FORECAST_ACTUAL_DIFFERENCE".equals(reason)) worker.getQueue().clear(); - pending.clear(); - LOG.info("Lingxing quantity capture stopped: {}", reason); - } - private synchronized void abortTransmission(String reason) { - worker.getQueue().clear(); pending.clear(); - if (!reason.equals(stopReason)) { - stopReason = reason; LOG.info("Lingxing quantity capture stopped: {}", reason); - } - } - public String stopReason() { return stopReason; } - @PreDestroy public void close() { stop("SHUTDOWN"); worker.shutdownNow(); } - private static String text(JsonNode data, String field, int max) { - String value = optionalText(data, field, max); - if (value == null || value.isBlank()) throw new IllegalArgumentException("Missing diagnostic identity"); - return value; - } - private static String optionalText(JsonNode data, String field, int max) { - JsonNode value = data.get(field); - if (value == null || value.isNull()) return null; - if (!value.isTextual() || value.textValue().length() > max) throw new IllegalArgumentException("Invalid diagnostic text"); - return value.textValue(); - } - private static BigDecimal quantity(JsonNode value) { - if (value == null || value.isNull() || value.isTextual() && value.textValue().isBlank()) return null; - if (!value.isNumber()) throw new IllegalArgumentException("Invalid diagnostic quantity"); - BigDecimal result = value.decimalValue(); - if (result.precision() > 30 || Math.abs(result.scale()) > 10) throw new IllegalArgumentException("Quantity limit"); - return result; - } - private record Key(String warehouse, String customer, String order) { } - private record Quantities(BigDecimal list, BigDecimal delivery, BigDecimal actual) { - Quantities add(Quantities other) { return new Quantities(sum(list, other.list), sum(delivery, other.delivery), sum(actual, other.actual)); } - private static BigDecimal sum(BigDecimal a, BigDecimal b) { return a == null || b == null ? null : a.add(b); } - } - private record Sample(Key key, boolean detail, Observation observation, Map quantities, int rowCount) { } - private static final class Pending { - private Sample list, detail; - } + private static final int MAX_BYTES=2097152,MAX_ROWS=2000,MAX_QUEUE=100; + private final LingxingQuantityCaptureProperties config;private final Sink sink;private final Clock clock; + private final String producer=UUID.randomUUID().toString(); + private final ScheduledExecutorService worker; + private final ReentrantLock intake=new ReentrantLock(); + private final ArrayDeque queue=new ArrayDeque<>(); + private final Set selected=new HashSet<>(),details=new HashSet<>(); + private final AtomicReference failure=new AtomicReference<>(); + private String observedRoot; + private volatile View grant;private volatile boolean closed;private int bytes,rows;private boolean attached; + @Autowired public LingxingQuantityCapture(LingxingQuantityCaptureProperties config,ObjectMapper mapper,ObjectProvider sink){this(config,mapper,sink.getIfAvailable(),Clock.systemUTC());} + LingxingQuantityCapture(LingxingQuantityCaptureProperties config,ObjectMapper mapper,Sink sink,Clock clock){this(config,sink,clock,true);} + LingxingQuantityCapture(LingxingQuantityCaptureProperties config,Sink sink,Clock clock,boolean start){ + this.config=config;this.sink=sink;this.clock=clock; + worker=Executors.newSingleThreadScheduledExecutor(r->{Thread t=new Thread(r,"lingxing-quantity-capture");t.setDaemon(true);return t;}); + if(start&&config.enabled()&&sink!=null)worker.scheduleWithFixedDelay(this::pump,0,2,TimeUnit.SECONDS); + } + public void observe(JsonNode data,Operation operation,EgressContext context,Instant observedAt){ + View current=grant; + if(closed||!config.enabled()||failure.get()!=null||current==null||"TERMINAL".equals(current.state())||"DRAINING".equals(current.state())||!clock.instant().isBefore(current.expiresAt()))return; + if(context==null||!context.wellFormed()||!"BUSINESS".equals(String.valueOf(context.executionPurpose()))||!Objects.equals(config.sourceAccount(),context.sourceAccount()))return; + if(operation!=Operation.OUTBOUND_QUERY&&operation!=Operation.OUTBOUND_DETAIL)return; + if(current.rootTaskId()!=null&&!current.rootTaskId().equals(context.rootTaskId())||Objects.equals(current.excludedRootTaskId(),context.rootTaskId()))return; + if(!intake.tryLock()){failure.compareAndSet(null,"OBSERVER_FAILED");return;} + try{ + if(current!=grant){failure.compareAndSet(null,"OBSERVER_FAILED");return;} + if(operation==Operation.OUTBOUND_DETAIL)snapshot(data,true,context,observedAt,current); + else {JsonNode records=data.path("records");if(!records.isArray()||records.size()>100){failure.compareAndSet(null,"OBSERVER_FAILED");return;} + for(JsonNode row:records){if(failure.get()!=null)break;snapshot(row,false,context,observedAt,current);}} + }catch(RuntimeException bad){failure.compareAndSet(null,"OBSERVER_FAILED");}finally{intake.unlock();} + } + private void snapshot(JsonNode data,boolean detail,EgressContext c,Instant time,View g){ + String warehouse=text(data,"whCode",64),owner=text(data,"customerCode",128),order=text(data,"deliveryNo",128); + if(!g.warehouseCodes().contains(warehouse)||!("*".equals(g.ownerCode())||g.ownerCode().equals(owner)))return; + if(observedRoot!=null&&!observedRoot.equals(c.rootTaskId()))return; + String key=key(c.rootTaskId(),warehouse,owner,order); + if(detail&&(!selected.contains(key)||details.contains(key)))return; + if(!detail&&(selected.contains(key)||selected.size()>=100))return; + JsonNode items=data.path(detail?"skuVOList":"productList");boolean complete=items.isArray()&&!items.isEmpty(); + if(items.size()>200){failure.compareAndSet(null,"OBSERVER_FAILED");return;} + // Conservative UTF-8 bound: identifiers <=128 chars, quantity <=38 digits, bounded metadata. + int retained=4096+items.size()*1024; + if(queue.size()>=MAX_QUEUE||rows+items.size()>MAX_ROWS||bytes+retained>MAX_BYTES){failure.compareAndSet(null,"QUEUE_FULL");return;} + var quantities=new TreeMap(); + if(items.isArray())for(JsonNode item:items){String sku=text(item,"productSku",128); + Sku value=new Sku(sku,detail?null:quantity(item.get("qty")),detail?quantity(item.get("deliveryQty")):null,detail?quantity(item.get("actualDeliveryQty")):null); + quantities.merge(sku,value,(a,b)->new Sku(sku,sum(a.listQty(),b.listQty()),sum(a.forecastQty(),b.forecastQty()),sum(a.actualQty(),b.actualQty()))); + } + String status=data.path("status").isIntegralNumber()?data.path("status").asText():null; + if(status!=null&&status.length()>32)throw new IllegalArgumentException(); + Event event=new Event(g.diagnosticSessionId(),c.sourceAccount(),producer,c.rootTaskId(),c.sessionId(),detail?"OUTBOUND_DETAIL":"OUTBOUND_QUERY",warehouse,owner,order,detail?"DETAIL":"LIST",new Observation(status,optional(data,"outboundTime",64),time,c.attemptId(),complete),List.copyOf(quantities.values())); + queue.addLast(new Queued(event,retained,items.size()));bytes+=retained;rows+=items.size(); + if(detail)details.add(key);else {selected.add(key);observedRoot=c.rootTaskId();} + } + /** Single worker owns HTTP; status(drained) occurs only after previously queued observations finish. */ + void pump(){ + if(closed||!config.enabled()||sink==null)return; + try{ + if(grant==null||"TERMINAL".equals(grant.state())){ + View next=sink.status(new Status(config.sourceAccount(),producer,!attached,false));attached=true; + if(next==null)return; + if(grant==null||!grant.diagnosticSessionId().equals(next.diagnosticSessionId()))reset(next);else grant=next; + } + if(grant==null||"TERMINAL".equals(grant.state()))return; + if(failure.get()!=null){grant=sink.stop(new Stop(grant.diagnosticSessionId(),config.sourceAccount(),producer,failure.get()));clearQueue();return;} + while(true){ + if(failure.get()!=null)break; + Queued next=poll();if(next==null)break; + try{ + if(!clock.instant().isBefore(grant.expiresAt())){failure.compareAndSet(null,"INCOMPLETE");break;} + Receipt receipt=sink.observe(next.event());if(receipt==null||receipt.session()==null)throw new IllegalStateException(); + grant=receipt.session(); + if("TERMINAL".equals(grant.state())){clearQueue();break;} + }finally{release(next);} + } + if(failure.get()!=null&&!"TERMINAL".equals(grant.state())){grant=sink.stop(new Stop(grant.diagnosticSessionId(),config.sourceAccount(),producer,failure.get()));clearQueue();return;} + if(!"TERMINAL".equals(grant.state())){ + View next=sink.status(new Status(config.sourceAccount(),producer,false,false));if(next==null)throw new IllegalStateException();grant=next; + if("TERMINAL".equals(next.state()))clearQueue(); + // After DRAINING is visible to intake, a later pump drains its tail before acknowledging. + if("DRAINING".equals(next.state())&&empty())grant=sink.status(new Status(config.sourceAccount(),producer,false,true)); + } + }catch(Exception failed){failure.compareAndSet(null,"INCOMPLETE");} + } + private Queued poll(){intake.lock();try{return queue.pollFirst();}finally{intake.unlock();}} + private void release(Queued q){intake.lock();try{bytes-=q.bytes();rows-=q.rows();}finally{intake.unlock();}} + private boolean empty(){intake.lock();try{return queue.isEmpty();}finally{intake.unlock();}} + private void clearQueue(){intake.lock();try{for(Queued q:queue){bytes-=q.bytes();rows-=q.rows();}queue.clear();}finally{intake.unlock();}} + private void reset(View v){intake.lock();try{queue.clear();bytes=0;rows=0;selected.clear();details.clear();failure.set(null);observedRoot=v.rootTaskId(); + for(Key k:v.selected()){String key=key(v.rootTaskId(),k.warehouseCode(),k.ownerCode(),k.orderNo());selected.add(key);if(k.paired())details.add(key);}grant=v; + }finally{intake.unlock();}} + private static String key(String root,String warehouse,String owner,String order){return root+"\u001f"+warehouse+"\u001f"+owner+"\u001f"+order;} + public String stopReason(){return failure.get()!=null?failure.get():grant==null?null:grant.stopReason();} + @PreDestroy public void close(){closed=true;worker.shutdownNow();clearQueue();} + private static String text(JsonNode n,String field,int max){String s=optional(n,field,max);if(s==null||s.isBlank())throw new IllegalArgumentException();return s;} + private static String optional(JsonNode n,String field,int max){JsonNode v=n.get(field);if(v==null||v.isNull())return null;if(!v.isTextual()||v.textValue().length()>max||v.textValue().codePoints().anyMatch(Character::isISOControl))throw new IllegalArgumentException();return v.textValue();} + private static BigDecimal quantity(JsonNode n){if(n==null||!n.isNumber())return null;BigDecimal value=n.decimalValue();return value.signum()<0||value.scale()>6||value.precision()-value.scale()>32?null:value.stripTrailingZeros();} + private static BigDecimal sum(BigDecimal a,BigDecimal b){return a==null||b==null?null:a.add(b);} + private record Queued(Event event,int bytes,int rows){} } diff --git a/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingQuantityEvidence.java b/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingQuantityEvidence.java index 0e871d723..4b3601e0d 100644 --- a/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingQuantityEvidence.java +++ b/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingQuantityEvidence.java @@ -1,34 +1,17 @@ package com.baiyue.integration.provider.lingxing; - import java.math.BigDecimal; import java.time.Instant; import java.util.List; - -/** Whitelist only; never includes source payload, addresses, labels or credentials. */ -public record LingxingQuantityEvidence(String diagnosticSessionId, String sourceAccount, String rootTaskId, - String collectionSessionId, String warehouseCode, String ownerCode, String orderNo, - Observation listObservation, Observation detailObservation, List skus) { - public record Observation(String status, String outboundTime, Instant observedAt, String attemptId, - boolean skuSetComplete) { } - public record Line(String sku, boolean listPresent, boolean detailPresent, BigDecimal listQty, - BigDecimal forecastQty, BigDecimal actualQty) { } - public boolean skuSetEqual() { return skus.stream().allMatch(line -> line.listPresent() && line.detailPresent()); } - public boolean forecastActualDifference() { return skus.stream().anyMatch(line -> line.forecastQty() != null - && line.actualQty() != null && line.forecastQty().compareTo(line.actualQty()) != 0); } - /** Receiver must durably enforce window, deduplication and cumulative bounds across restarts. */ - public interface Sink { void accept(LingxingQuantityEvidence evidence); } - /** Normal durable receiver decision, including an accepted final order; not a transport failure. */ - public static final class ReceiverStopped extends RuntimeException { - private final String reason; - private final boolean orderStored; - public ReceiverStopped(String reason, boolean orderStored) { - super("Diagnostic receiver stopped"); - if (reason == null || !reason.matches("[A-Z][A-Z0-9_]{0,63}")) { - throw new IllegalArgumentException("Invalid diagnostic stop reason"); - } - this.reason = reason; this.orderStored = orderStored; - } - public String reason() { return reason; } - public boolean orderStored() { return orderStored; } - } +/** Internal diagnostic whitelist; excludes raw payloads, addresses and supplier credentials. */ +public final class LingxingQuantityEvidence { + private LingxingQuantityEvidence(){} + public record Status(String sourceAccount,String producerId,boolean attach,boolean drained){} + public record Stop(String diagnosticSessionId,String sourceAccount,String producerId,String reason){} + public record Observation(String status,String outboundTime,Instant observedAt,String attemptId,Boolean skuSetComplete){} + public record Sku(String sku,BigDecimal listQty,BigDecimal forecastQty,BigDecimal actualQty){} + public record Event(String diagnosticSessionId,String sourceAccount,String producerId,String rootTaskId,String collectionSessionId,String operation,String warehouseCode,String ownerCode,String orderNo,String side,Observation observation,List skus){} + public record Key(String warehouseCode,String ownerCode,String orderNo,boolean paired){} + public record View(String diagnosticSessionId,String sourceAccount,String rootTaskId,String collectionSessionId,String state,String stopReason,Instant startsAt,Instant expiresAt,Instant bindBefore,List warehouseCodes,String ownerCode,String excludedRootTaskId,int candidates,int pairedOrders,int retainedRows,long retainedBytes,boolean deliveryUncertain,List selected){} + public record Receipt(String outcome,String comparison,View session){} + public interface Sink {View status(Status request);Receipt observe(Event event);View stop(Stop request);} } diff --git a/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingQuantityHttpSink.java b/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingQuantityHttpSink.java index 0a5571b22..897baff0e 100644 --- a/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingQuantityHttpSink.java +++ b/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingQuantityHttpSink.java @@ -1,7 +1,6 @@ package com.baiyue.integration.provider.lingxing; - +import static com.baiyue.integration.provider.lingxing.LingxingQuantityEvidence.*; import com.baiyue.integration.config.LingxingEgressProperties; -import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; @@ -9,49 +8,43 @@ import java.net.URI; import java.net.http.*; import java.nio.charset.StandardCharsets; import java.time.Duration; -import java.util.Set; -/** Async worker calls this once per paired order. No retry and no supplier requests. */ +/** Invoked only by the bounded observer worker. Never retries or contacts the supplier. */ @Component -public final class LingxingQuantityHttpSink implements LingxingQuantityEvidence.Sink { - private static final String PATH = "/internal/lastleg/self-operated-analytics/quantity-diagnostics/orders"; - private final URI baseUri; - private final ObjectMapper mapper; - private final String token; - private final HttpClient client = HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(2)) - .followRedirects(HttpClient.Redirect.NEVER).build(); - public LingxingQuantityHttpSink(LingxingEgressProperties config, ObjectMapper mapper, - @Value("${baiyue.security.internal-service-token:}") String token) { - this.baseUri = config.budgetAuthorityBaseUri(); - this.mapper = mapper.copy().disable(com.fasterxml.jackson.databind.SerializationFeature.WRITE_DATES_AS_TIMESTAMPS); - this.token = token; - } - @Override public void accept(LingxingQuantityEvidence evidence) { - try { - if (baseUri == null || token == null || token.getBytes(StandardCharsets.UTF_8).length < 32) throw unavailable(); - byte[] body = mapper.writeValueAsBytes(evidence); - if (body.length > 256 * 1024 || evidence.skus().size() > 200) throw unavailable(); - HttpRequest request = HttpRequest.newBuilder(baseUri.resolve(PATH)).timeout(Duration.ofSeconds(3)) - .header("X-Internal-Service-Token", token).header("Content-Type", "application/json; charset=UTF-8") - .POST(HttpRequest.BodyPublishers.ofByteArray(body)).build(); - HttpResponse response = client.send(request, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8)); - if (response.statusCode() != 200 || response.body().length() > 16384) throw unavailable(); - JsonNode root = mapper.readTree(response.body()); - if (root == null || !"200".equals(root.path("code").textValue())) throw unavailable(); - String outcome = root.path("data").path("outcome").asText(); - if (!Set.of("ACCEPTED", "DUPLICATE", "STOPPED").contains(outcome)) throw unavailable(); - // Receiver may accept the final order and atomically close its durable session. - JsonNode reason = root.path("data").path("stopReason"); - if (!reason.isMissingNode() && !reason.isNull()) { - if (!reason.isTextual() || !reason.textValue().matches("[A-Z][A-Z0-9_]{0,63}")) throw unavailable(); - throw new LingxingQuantityEvidence.ReceiverStopped(reason.textValue(), !"STOPPED".equals(outcome)); - } - if ("STOPPED".equals(outcome)) throw unavailable(); - } catch (LingxingQuantityEvidence.ReceiverStopped normalStop) { - throw normalStop; - } catch (InterruptedException interrupted) { - Thread.currentThread().interrupt(); throw unavailable(); - } catch (Exception failed) { throw unavailable(); } - } - private static IllegalStateException unavailable() { return new IllegalStateException("DIAGNOSTIC_RECEIVER_UNAVAILABLE"); } +public final class LingxingQuantityHttpSink implements Sink { + private static final String PATH="/internal/lastleg/self-operated-analytics/quantity-diagnostics/"; + private final URI baseUri;private final ObjectMapper json;private final String token; + private final HttpClient client=HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(2)).followRedirects(HttpClient.Redirect.NEVER).build(); + public LingxingQuantityHttpSink(LingxingEgressProperties config,ObjectMapper json,@Value("${baiyue.security.internal-service-token:}") String token){this.baseUri=config.budgetAuthorityBaseUri();this.json=json.copy().disable(com.fasterxml.jackson.databind.SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);this.token=token;} + public View status(Status request){return call("status",request,View.class);} + public Receipt observe(Event event){return call("observe",event,Receipt.class);} + public View stop(Stop request){return call("stop",request,View.class);} + private T call(String path,Object payload,Class type){ + try{ + if(baseUri==null||token==null||token.getBytes(StandardCharsets.UTF_8).length<32)throw unavailable(); + byte[] body=json.writeValueAsBytes(payload);if(body.length>262144)throw unavailable(); + HttpRequest request=HttpRequest.newBuilder(baseUri.resolve(PATH+path)).timeout(Duration.ofSeconds(3)).header("X-Internal-Service-Token",token).header("Content-Type","application/json; charset=UTF-8").POST(HttpRequest.BodyPublishers.ofByteArray(body)).build(); + // Limit while receiving, not after allocating an arbitrary response String. + HttpResponse response=client.send(request,info->new LimitedBody(262144)); + if(response.statusCode()!=200)throw unavailable(); + var root=json.readTree(response.body());if(root==null||!"200".equals(root.path("code").asText()))throw unavailable(); + if(root.path("data").isNull())return null; + return json.treeToValue(root.path("data"),type); + }catch(InterruptedException e){Thread.currentThread().interrupt();throw unavailable();} + catch(Exception e){throw unavailable();} + } + private static IllegalStateException unavailable(){return new IllegalStateException("DIAGNOSTIC_RECEIVER_UNAVAILABLE");} + static final class LimitedBody implements HttpResponse.BodySubscriber { + private final int limit;private final java.io.ByteArrayOutputStream out=new java.io.ByteArrayOutputStream(); + private final java.util.concurrent.CompletableFuture result=new java.util.concurrent.CompletableFuture<>(); + private java.util.concurrent.Flow.Subscription subscription; + LimitedBody(int limit){this.limit=limit;} + public java.util.concurrent.CompletionStage getBody(){return result;} + public void onSubscribe(java.util.concurrent.Flow.Subscription s){subscription=s;s.request(1);} + public void onNext(java.util.List buffers){ + for(var buffer:buffers){if(buffer.remaining()>limit-out.size()){subscription.cancel();result.completeExceptionally(unavailable());return;}byte[] bytes=new byte[buffer.remaining()];buffer.get(bytes);out.writeBytes(bytes);}subscription.request(1); + } + public void onError(Throwable error){result.completeExceptionally(error);} + public void onComplete(){result.complete(out.toByteArray());} + } } diff --git a/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingWmsClient.java b/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingWmsClient.java index 33f3d4373..e87136021 100644 --- a/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingWmsClient.java +++ b/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingWmsClient.java @@ -192,9 +192,9 @@ public final class LingxingWmsClient { } T result = projection.apply(json, handle.metadata()); if (quantityCapture != null && (operation == Operation.OUTBOUND_QUERY - || operation == Operation.OUTBOUND_LOCATE || operation == Operation.OUTBOUND_DETAIL)) { + || operation == Operation.OUTBOUND_DETAIL)) { // Diagnostics must never alter the normal projection or the egress completion result. - try { quantityCapture.observe(json, operation == Operation.OUTBOUND_DETAIL, context, clock.instant()); } + try { quantityCapture.observe(json, operation, context, clock.instant()); } catch (RuntimeException ignored) { /* Passive diagnostic failure is isolated. */ } } return result; diff --git a/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingQuantityCaptureTest.java b/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingQuantityCaptureTest.java index 7624cd4d5..d6b50bd31 100644 --- a/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingQuantityCaptureTest.java +++ b/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingQuantityCaptureTest.java @@ -1,146 +1,50 @@ package com.baiyue.integration.provider.lingxing; - +import static com.baiyue.integration.provider.lingxing.LingxingQuantityEvidence.*; +import static org.assertj.core.api.Assertions.*; import com.baiyue.integration.config.LingxingQuantityCaptureProperties; import com.baiyue.integration.dto.lingxing.LingxingEgressDtos.*; import com.fasterxml.jackson.databind.ObjectMapper; -import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.ValueSource; import java.time.*; +import java.util.*; import java.util.concurrent.*; -import static org.assertj.core.api.Assertions.*; - +import org.junit.jupiter.api.Test; class LingxingQuantityCaptureTest { - private static final Instant NOW = Instant.parse("2026-10-13T01:00:00Z"); - private final ObjectMapper mapper = new ObjectMapper().registerModule(new JavaTimeModule()); - private final BlockingQueue saved = new LinkedBlockingQueue<>(); - private EgressContext context(String session) { return new EgressContext("account", CollectionMode.BUSINESS, - "root", session, "attempt", "grant"); } - private LingxingQuantityCapture capture(Instant expiry, LingxingQuantityEvidence.Sink sink) { - return new LingxingQuantityCapture(new LingxingQuantityCaptureProperties(true, "capture", "account", - "root", "session", "*", "*", NOW.minusSeconds(1), expiry), mapper, sink, Clock.fixed(NOW, ZoneOffset.UTC)); - } - private void observe(LingxingQuantityCapture c, boolean detail, String items, String session, int status) throws Exception { - String order = "{\"whCode\":\"W\",\"customerCode\":\"C\",\"deliveryNo\":\"D\",\"status\":" + status - + ",\"" + (detail ? "skuVOList" : "productList") + "\":[" + items + "]}"; - c.observe(mapper.readTree(detail ? order : "{\"records\":[" + order + "]}"), detail, context(session), NOW); - } - @ParameterizedTest @ValueSource(ints = {10, 8}) - void preservesBothSemanticOutcomesAndStopsAtDifference(int listQty) throws Exception { - try (var c = new CloseableCapture(capture(NOW.plusSeconds(60), saved::add))) { - observe(c.value, false, "{\"productSku\":\"S\",\"qty\":" + listQty + "}", "session", 100); - observe(c.value, true, "{\"productSku\":\"S\",\"deliveryQty\":10,\"actualDeliveryQty\":8}", "session", 100); - var e = saved.poll(3, TimeUnit.SECONDS); - assertThat(e).isNotNull(); assertThat(e.forecastActualDifference()).isTrue(); - assertThat(e.skus().get(0).listQty()).isEqualByComparingTo(Integer.toString(listQty)); - assertThat(e.skuSetEqual()).isTrue(); - awaitStop(c.value, "FORECAST_ACTUAL_DIFFERENCE"); - } - } - @Test void duplicateDecimalsUnknownsAndSkuMismatchRemainVisibleForCancelledOrder() throws Exception { - try (var c = new CloseableCapture(capture(NOW.plusSeconds(60), saved::add))) { - observe(c.value, false, "{\"productSku\":\"S\",\"qty\":0.1},{\"productSku\":\"S\",\"qty\":0.2}," - + "{\"productSku\":\"LIST_ONLY\",\"qty\":1}", "session", 99); - observe(c.value, true, "{\"productSku\":\"S\",\"deliveryQty\":1,\"actualDeliveryQty\":null}," - + "{\"productSku\":\"S\",\"deliveryQty\":2,\"actualDeliveryQty\":1}", "session", 99); - var e = saved.poll(3, TimeUnit.SECONDS); - assertThat(e).isNotNull(); assertThat(e.skuSetEqual()).isFalse(); - var line = e.skus().stream().filter(v -> v.sku().equals("S")).findFirst().orElseThrow(); - assertThat(line.listQty()).isEqualByComparingTo("0.3"); - assertThat(line.forecastQty()).isEqualByComparingTo("3"); assertThat(line.actualQty()).isNull(); - assertThat(e.detailObservation().status()).isEqualTo("99"); assertThat(e.detailObservation().outboundTime()).isNull(); - assertThat(e.forecastActualDifference()).isFalse(); - } - } - @Test void crossSessionCannotPairAndExpiredCaptureDoesNotWrite() throws Exception { - try (var c = new CloseableCapture(capture(NOW.plusSeconds(60), saved::add))) { - observe(c.value, false, "{\"productSku\":\"S\",\"qty\":1}", "other", 100); - observe(c.value, true, "{\"productSku\":\"S\",\"deliveryQty\":1,\"actualDeliveryQty\":1}", "session", 100); - assertThat(saved.poll(100, TimeUnit.MILLISECONDS)).isNull(); - } - try (var c = new CloseableCapture(capture(NOW, saved::add))) { - observe(c.value, false, "{\"productSku\":\"S\",\"qty\":1}", "session", 100); - assertThat(c.value.stopReason()).isEqualTo("EXPIRED"); assertThat(saved).isEmpty(); - } - } - @Test void sinkFailureStopsOnlyDiagnosticAndNeverEscapesCaller() throws Exception { - try (var c = new CloseableCapture(capture(NOW.plusSeconds(60), evidence -> { throw new IllegalStateException("private"); }))) { - observe(c.value, false, "{\"productSku\":\"S\",\"qty\":1}", "session", 100); - observe(c.value, true, "{\"productSku\":\"S\",\"deliveryQty\":1,\"actualDeliveryQty\":1}", "session", 100); - awaitStop(c.value, "DIAGNOSTIC_FAILED"); - } - } - @Test void hundredCompleteOrdersStopsFurtherCapture() throws Exception { - try (var c = new CloseableCapture(capture(NOW.plusSeconds(60), saved::add))) { - for (int i = 0; i < 100; i++) { - String head = "\"whCode\":\"W\",\"customerCode\":\"C\",\"deliveryNo\":\"D" + i + "\",\"status\":100,"; - c.value.observe(mapper.readTree("{\"records\":[{" + head + "\"productList\":[{\"productSku\":\"S\",\"qty\":1}]}]}"), false, context("session"), NOW); - c.value.observe(mapper.readTree("{" + head + "\"skuVOList\":[{\"productSku\":\"S\",\"deliveryQty\":1,\"actualDeliveryQty\":1}]}"), true, context("session"), NOW); - assertThat(saved.poll(3, TimeUnit.SECONDS)).isNotNull(); - } - awaitStop(c.value, "ORDER_LIMIT"); - observe(c.value, false, "{\"productSku\":\"S\",\"qty\":1}", "session", 100); - assertThat(saved).isEmpty(); - } - } - @Test void thousandRowPageSelectsHundredAndSlowSinkDoesNotBlockDetails() throws Exception { - CountDownLatch entered = new CountDownLatch(1), release = new CountDownLatch(1); - try (var c = new CloseableCapture(capture(NOW.plusSeconds(60), evidence -> { - entered.countDown(); - try { release.await(3, TimeUnit.SECONDS); } catch (InterruptedException error) { Thread.currentThread().interrupt(); } - saved.add(evidence); - }))) { - var page = mapper.createObjectNode(); var records = page.putArray("records"); - for (int i = 0; i < 1000; i++) { - var order = records.addObject().put("whCode", "W").put("customerCode", "C").put("deliveryNo", "D" + i).put("status", 100); - order.putArray("productList").addObject().put("productSku", "S").put("qty", 1); - } - c.value.observe(page, false, context("session"), NOW); - assertThat(c.value.stopReason()).isNull(); - // Unselected details arrive first: they must not consume queue or evidence capacity. - for (int i = 999; i >= 0; i--) { - var detail = mapper.createObjectNode().put("whCode", "W").put("customerCode", "C").put("deliveryNo", "D" + i).put("status", 100); - detail.putArray("skuVOList").addObject().put("productSku", "S").put("deliveryQty", 1).put("actualDeliveryQty", 1); - c.value.observe(detail, true, context("session"), NOW); - } - assertThat(entered.await(3, TimeUnit.SECONDS)).isTrue(); - assertThat(c.value.stopReason()).isEqualTo("ORDER_LIMIT"); - release.countDown(); - for (int i = 0; i < 100; i++) assertThat(saved.poll(3, TimeUnit.SECONDS)).isNotNull(); - assertThat(saved).isEmpty(); - } finally { release.countDown(); } - } - private static void awaitStop(LingxingQuantityCapture capture, String reason) throws Exception { - long end = System.nanoTime() + TimeUnit.SECONDS.toNanos(3); - while (capture.stopReason() == null && System.nanoTime() < end) Thread.sleep(5); - assertThat(capture.stopReason()).isEqualTo(reason); - } - @ParameterizedTest @ValueSource(strings = {"FORECAST_ACTUAL_DIFFERENCE", "LIMIT_REACHED"}) - void savedFinalOrderRetainsNormalReceiverReasonAndDiscardsQueuedOrders(String reason) throws Exception { - CountDownLatch entered = new CountDownLatch(1), release = new CountDownLatch(1); - try (var c = new CloseableCapture(capture(NOW.plusSeconds(60), evidence -> { - entered.countDown(); - try { release.await(3, TimeUnit.SECONDS); } catch (InterruptedException error) { Thread.currentThread().interrupt(); } - saved.add(evidence); // Receiver persisted the last accepted order before returning its stop decision. - throw new LingxingQuantityEvidence.ReceiverStopped(reason, true); - }))) { - for (int i = 0; i < 3; i++) { - var order = mapper.createObjectNode().put("whCode", "W").put("customerCode", "C").put("deliveryNo", "D" + i).put("status", 100); - order.putArray("productList").addObject().put("productSku", "S").put("qty", 1); - var page = mapper.createObjectNode(); page.putArray("records").add(order); - c.value.observe(page, false, context("session"), NOW); - order.putArray("skuVOList").addObject().put("productSku", "S").put("deliveryQty", 1).put("actualDeliveryQty", 1); - c.value.observe(order, true, context("session"), NOW); - } - assertThat(entered.await(3, TimeUnit.SECONDS)).isTrue(); - release.countDown(); awaitStop(c.value, reason); - assertThat(saved).hasSize(1); - assertThat(saved.remove().orderNo()).isEqualTo("D0"); - assertThat(saved.poll(100, TimeUnit.MILLISECONDS)).isNull(); - } finally { release.countDown(); } - } - private record CloseableCapture(LingxingQuantityCapture value) implements AutoCloseable { - @Override public void close() { value.close(); } - } + final Instant now=Instant.parse("2026-10-14T00:00:00Z");final ObjectMapper json=new ObjectMapper(); + final EgressContext context=new EgressContext("account",CollectionMode.BUSINESS,"2","selfop-2","attempt","grant"); + View view(){return new View("diag","account",null,null,"ARMED",null,now,now.plusSeconds(36000),now.plusSeconds(21600),List.of("W"),"*","1",0,0,0,0,false,List.of());} + com.fasterxml.jackson.databind.JsonNode list(String id)throws Exception{return json.readTree("{\"records\":[{\"whCode\":\"W\",\"customerCode\":\"C\",\"deliveryNo\":\""+id+"\",\"productList\":[{\"productSku\":\"S\",\"qty\":10}]}]}");} + class Fake implements Sink {View state=view();List received=new ArrayList<>();String stopped;public View status(Status s){return state;}public Receipt observe(Event e){received.add(e);return new Receipt("ACCEPTED",null,state);}public View stop(Stop s){stopped=s.reason();return state;}} + LingxingQuantityCapture capture(Sink sink){return new LingxingQuantityCapture(new LingxingQuantityCaptureProperties(true,"account"),sink,Clock.fixed(now,ZoneOffset.UTC),false);} + @Test void observationDoesNotCallReceiverAndLocateNeverBinds()throws Exception{ + Fake sink=new Fake();try(var c=new TestClose(capture(sink))){c.value.pump();c.value.observe(list("a"),Operation.OUTBOUND_LOCATE,context,now);c.value.pump();assertThat(sink.received).isEmpty(); + c.value.observe(list("a"),Operation.OUTBOUND_QUERY,context,now);assertThat(sink.received).isEmpty();c.value.pump();assertThat(sink.received).hasSize(1);assertThat(sink.received.get(0).operation()).isEqualTo("OUTBOUND_QUERY");} + } + @Test void slowReceiverNeverBlocksNormalObservationAndFullQueueStopsOnlyDiagnosis()throws Exception{ + CountDownLatch entered=new CountDownLatch(1),release=new CountDownLatch(1);Fake sink=new Fake(){public Receipt observe(Event e){entered.countDown();try{release.await(3,TimeUnit.SECONDS);}catch(InterruptedException x){Thread.currentThread().interrupt();}return super.observe(e);}}; + var c=capture(sink);c.pump();c.observe(list("first"),Operation.OUTBOUND_QUERY,context,now);var worker=Executors.newSingleThreadExecutor();try{var pending=worker.submit(c::pump);assertThat(entered.await(1,TimeUnit.SECONDS)).isTrue(); + assertThatCode(()->{for(int i=0;i<100;i++)c.observe(list("o"+i),Operation.OUTBOUND_QUERY,context,now);}).doesNotThrowAnyException(); + release.countDown();pending.get(5,TimeUnit.SECONDS); + }finally{release.countDown();worker.shutdownNow();c.close();} + } + @Test void failureIsolatedAndNoRearmOfSameGrant()throws Exception{ + Fake sink=new Fake(){public Receipt observe(Event e){throw new IllegalStateException("receiver down");}};var c=capture(sink);try{c.pump();c.observe(list("a"),Operation.OUTBOUND_QUERY,context,now);assertThatCode(c::pump).doesNotThrowAnyException();assertThat(c.stopReason()).isEqualTo("INCOMPLETE");c.pump();assertThat(sink.stopped).isEqualTo("INCOMPLETE");}finally{c.close();} + } + @Test void selectedCandidatesRecoveredWithoutReobservingList()throws Exception{ + Fake sink=new Fake();sink.state=new View("diag","account","2","selfop-2","BOUND",null,now,now.plusSeconds(14400),now.plusSeconds(21600),List.of("W"),"*","1",1,0,1,100,true,List.of(new Key("W","C","a",false))); + var c=capture(sink);try{c.pump();c.observe(list("a"),Operation.OUTBOUND_QUERY,context,now);c.observe(json.readTree("{\"whCode\":\"W\",\"customerCode\":\"C\",\"deliveryNo\":\"a\",\"skuVOList\":[{\"productSku\":\"S\",\"deliveryQty\":10,\"actualDeliveryQty\":10}]}"),Operation.OUTBOUND_DETAIL,context,now);c.pump();assertThat(sink.received).hasSize(1);assertThat(sink.received.get(0).side()).isEqualTo("DETAIL");}finally{c.close();} + } + @Test void malformedOrOversizedSnapshotsAreIsolated()throws Exception{ + Fake sink=new Fake();var c=capture(sink);try{c.pump();var data=json.createObjectNode();var records=data.putArray("records");for(int i=0;i<101;i++)records.addObject();assertThatCode(()->c.observe(data,Operation.OUTBOUND_QUERY,context,now)).doesNotThrowAnyException();c.pump();assertThat(sink.stopped).isEqualTo("OBSERVER_FAILED");assertThat(sink.received).isEmpty();}finally{c.close();} + } + @Test void queueIsBoundedAndNoProducerWaitsForIt()throws Exception{ + Fake sink=new Fake();var c=capture(sink);try{c.pump();for(int i=0;i<100;i++)c.observe(list("o"+i),Operation.OUTBOUND_QUERY,context,now); + var detail=json.readTree("{\"whCode\":\"W\",\"customerCode\":\"C\",\"deliveryNo\":\"o0\",\"skuVOList\":[{\"productSku\":\"S\",\"deliveryQty\":10,\"actualDeliveryQty\":10}]}"); + c.observe(detail,Operation.OUTBOUND_DETAIL,context,now);assertThat(c.stopReason()).isEqualTo("QUEUE_FULL");c.pump();assertThat(sink.stopped).isEqualTo("QUEUE_FULL");assertThat(sink.received).isEmpty(); + }finally{c.close();} + } + @Test void diagnosticLockContentionNeverWaitsOnCollector()throws Exception{ + Fake sink=new Fake();var c=capture(sink);c.pump();var field=LingxingQuantityCapture.class.getDeclaredField("intake");field.setAccessible(true);var lock=(java.util.concurrent.locks.ReentrantLock)field.get(c); + var pool=Executors.newSingleThreadExecutor();lock.lock();try{var observation=pool.submit(()->{try{c.observe(list("a"),Operation.OUTBOUND_QUERY,context,now);}catch(Exception e){throw new RuntimeException(e);}});observation.get(500,TimeUnit.MILLISECONDS);assertThat(c.stopReason()).isEqualTo("OBSERVER_FAILED");}finally{lock.unlock();pool.shutdownNow();c.close();} + } + record TestClose(LingxingQuantityCapture value)implements AutoCloseable{public void close(){value.close();}} } diff --git a/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingQuantityHttpSinkTest.java b/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingQuantityHttpSinkTest.java index 6fb44e246..2592d581b 100644 --- a/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingQuantityHttpSinkTest.java +++ b/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingQuantityHttpSinkTest.java @@ -1,65 +1,32 @@ package com.baiyue.integration.provider.lingxing; - -import com.baiyue.integration.config.LingxingEgressProperties; -import com.baiyue.integration.dto.lingxing.LingxingEgressDtos.CollectionMode; -import com.fasterxml.jackson.databind.ObjectMapper; -import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; -import com.sun.net.httpserver.HttpServer; import org.junit.jupiter.api.Test; -import java.net.*; -import java.nio.charset.StandardCharsets; -import java.time.Instant; +import java.nio.ByteBuffer; import java.util.List; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.Flow; import static org.assertj.core.api.Assertions.*; - class LingxingQuantityHttpSinkTest { - @Test void exactContractAuthenticationAndNoRetryWhenReceiverStops() throws Exception { - var calls = new AtomicInteger(); var body = new AtomicReference(); var header = new AtomicReference(); - var outcome = new AtomicReference<>("ACCEPTED"); - var reason = new AtomicReference<>("null"); - HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); - server.createContext("/internal/lastleg/self-operated-analytics/quantity-diagnostics/orders", exchange -> { - calls.incrementAndGet(); body.set(new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8)); - header.set(exchange.getRequestHeaders().getFirst("X-Internal-Service-Token")); - byte[] response = ("{\"code\":\"200\",\"data\":{\"outcome\":\"" + outcome.get() + "\",\"stopReason\":" + reason.get() + "}}") - .getBytes(StandardCharsets.UTF_8); - exchange.sendResponseHeaders(200, response.length); exchange.getResponseBody().write(response); exchange.close(); - }); server.start(); - try { - var props = new LingxingEgressProperties(CollectionMode.OFF, "test", null, null, null, - URI.create("http://127.0.0.1:" + server.getAddress().getPort()), null, null, false); - var mapper = new ObjectMapper().registerModule(new JavaTimeModule()) - .disable(com.fasterxml.jackson.databind.SerializationFeature.WRITE_DATES_AS_TIMESTAMPS); - String token = "synthetic-internal-token-000000000000"; - var sink = new LingxingQuantityHttpSink(props, mapper, token); - var obs = new LingxingQuantityEvidence.Observation("99", null, Instant.parse("2026-10-13T00:00:00Z"), "attempt", true); - var evidence = new LingxingQuantityEvidence("diagnostic", "account", "root", "collection", "W", "C", "D", obs, obs, - List.of(new LingxingQuantityEvidence.Line("S", true, true, null, null, null))); - sink.accept(evidence); - var json = mapper.readTree(body.get()); - assertThat(header.get()).isEqualTo(token); assertThat(json.path("diagnosticSessionId").asText()).isEqualTo("diagnostic"); - assertThat(json.path("listObservation").path("observedAt").asText()).isEqualTo("2026-10-13T00:00:00Z"); - assertThat(json.path("skus").get(0).path("forecastQty").isNull()).isTrue(); - assertThat(json.has("skuSetEqual")).isFalse(); assertThat(json.has("forecastActualDifference")).isFalse(); - outcome.set("STOPPED"); assertThatThrownBy(() -> sink.accept(evidence)).isInstanceOf(IllegalStateException.class); - assertThat(calls.get()).isEqualTo(2); - for (String stopReason : List.of("FORECAST_ACTUAL_DIFFERENCE", "LIMIT_REACHED")) { - outcome.set("ACCEPTED"); reason.set("\"" + stopReason + "\""); - assertThatThrownBy(() -> sink.accept(evidence)).isInstanceOfSatisfying( - LingxingQuantityEvidence.ReceiverStopped.class, stopped -> { - assertThat(stopped.reason()).isEqualTo(stopReason); - assertThat(stopped.orderStored()).isTrue(); - }); - } - outcome.set("STOPPED"); reason.set("\"DISABLED\""); - assertThatThrownBy(() -> sink.accept(evidence)).isInstanceOfSatisfying( - LingxingQuantityEvidence.ReceiverStopped.class, stopped -> { - assertThat(stopped.reason()).isEqualTo("DISABLED"); - assertThat(stopped.orderStored()).isFalse(); - }); - assertThat(calls.get()).isEqualTo(5); - } finally { server.stop(0); } - } + @Test void internalWireContractAuthenticatesAndDoesNotRetry()throws Exception{ + var server=com.sun.net.httpserver.HttpServer.create(new java.net.InetSocketAddress("127.0.0.1",0),0);var calls=new java.util.concurrent.atomic.AtomicInteger();var header=new java.util.concurrent.atomic.AtomicReference();var path=new java.util.concurrent.atomic.AtomicReference(); + server.createContext("/internal/lastleg/self-operated-analytics/quantity-diagnostics/",exchange->{calls.incrementAndGet();header.set(exchange.getRequestHeaders().getFirst("X-Internal-Service-Token"));path.set(exchange.getRequestURI().getPath());exchange.getRequestBody().readAllBytes();byte[] payload="{\"code\":\"200\",\"data\":null}".getBytes(java.nio.charset.StandardCharsets.UTF_8);exchange.sendResponseHeaders(200,payload.length);exchange.getResponseBody().write(payload);exchange.close();});server.start(); + try{ + var props=new com.baiyue.integration.config.LingxingEgressProperties(com.baiyue.integration.dto.lingxing.LingxingEgressDtos.CollectionMode.OFF,"test",null,null,null,java.net.URI.create("http://127.0.0.1:"+server.getAddress().getPort()),null,null,false); + String token="synthetic-internal-token-000000000000";var sink=new LingxingQuantityHttpSink(props,new com.fasterxml.jackson.databind.ObjectMapper().findAndRegisterModules(),token); + assertThat(sink.status(new LingxingQuantityEvidence.Status("account","producer",true,false))).isNull();assertThat(header.get()).isEqualTo(token);assertThat(path.get()).endsWith("/status");assertThat(calls.get()).isEqualTo(1); + sink.stop(new LingxingQuantityEvidence.Stop("diag","account","producer","INCOMPLETE"));assertThat(path.get()).endsWith("/stop");assertThat(calls.get()).isEqualTo(2); + }finally{server.stop(0);} + } + @Test void bodyLimitCancelsBeforeRetainingOversizedResponse(){ + var body=new LingxingQuantityHttpSink.LimitedBody(16);boolean[] cancelled={false};body.onSubscribe(new Flow.Subscription(){public void request(long n){}public void cancel(){cancelled[0]=true;}}); + body.onNext(List.of(ByteBuffer.wrap(new byte[8])));body.onNext(List.of(ByteBuffer.wrap(new byte[9]))); + assertThat(cancelled[0]).isTrue();assertThat(body.getBody().toCompletableFuture()).isCompletedExceptionally(); + } + @Test void hundredFullUnicodeKeysFitTheBoundedResponse()throws Exception{ + var keys=new java.util.ArrayList();for(int i=0;i<100;i++)keys.add(new LingxingQuantityEvidence.Key("倉".repeat(64),"客".repeat(128),"単".repeat(125)+i,false)); + byte[] payload=new com.fasterxml.jackson.databind.ObjectMapper().writeValueAsBytes(java.util.Map.of("code","200","data",java.util.Map.of("selected",keys))); + assertThat(payload.length).isGreaterThan(65536).isLessThan(262144); + var body=new LingxingQuantityHttpSink.LimitedBody(262144);body.onSubscribe(new Flow.Subscription(){public void request(long n){}public void cancel(){}});body.onNext(List.of(ByteBuffer.wrap(payload)));body.onComplete();assertThat(body.getBody().toCompletableFuture().join()).isEqualTo(payload); + } + @Test void boundedSuccessfulBodyIsExact(){ + var body=new LingxingQuantityHttpSink.LimitedBody(16);body.onSubscribe(new Flow.Subscription(){public void request(long n){}public void cancel(){}});body.onNext(List.of(ByteBuffer.wrap(new byte[]{1,2})));body.onComplete();assertThat(body.getBody().toCompletableFuture().join()).containsExactly(1,2); + } } diff --git a/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingWmsClientTest.java b/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingWmsClientTest.java index 29a4dd64a..d5b1deda4 100644 --- a/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingWmsClientTest.java +++ b/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingWmsClientTest.java @@ -55,7 +55,7 @@ class LingxingWmsClientTest { @Test void observerFailureCannotChangeBusinessResponseOrIssueExtraRequests() { var observer = org.mockito.Mockito.mock(LingxingQuantityCapture.class); org.mockito.Mockito.doThrow(new IllegalStateException("diagnostic failure")).when(observer) - .observe(org.mockito.ArgumentMatchers.any(), org.mockito.ArgumentMatchers.anyBoolean(), + .observe(org.mockito.ArgumentMatchers.any(), org.mockito.ArgumentMatchers.any(Operation.class), org.mockito.ArgumentMatchers.any(), org.mockito.ArgumentMatchers.any()); var c = client(); c.setQuantityCapture(observer); assertThat(c.queryPage(query(1, 1), context()).records()).hasSize(1); @@ -83,7 +83,10 @@ class LingxingWmsClientTest { } @Test void cancelledLocateOmitsTimeAndStatusAndRejectsFuzzyMatch() throws Exception { response.set(page("current", "size", 1, 1, 1, 1, ORDER.replace("100", "99").replace("2026-10-01 01:00:00", ""))); - var result = client().locate(new LocateRequest("DO-DEMO", 1, 1), context()); + var observer = org.mockito.Mockito.mock(LingxingQuantityCapture.class); + var c = client(); c.setQuantityCapture(observer); + var result = c.locate(new LocateRequest("DO-DEMO", 1, 1), context()); + org.mockito.Mockito.verifyNoInteractions(observer); var data = mapper.readTree(body.get()).path("data"); assertThat(data.size()).isEqualTo(3); assertThat(data.has("status")).isFalse(); assertThat(data.has("startTime")).isFalse(); diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticContract.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticContract.java index 832548d4e..0380816ca 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticContract.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticContract.java @@ -1,22 +1,16 @@ package com.baiyue.lastleg.selfoperated.analytics.diagnostic; - import java.math.BigDecimal; import java.time.Instant; import java.util.List; - -/** Diagnostic evidence only; never consumed by sales publication. */ public final class QuantityDiagnosticContract { - private QuantityDiagnosticContract() { } - public record Order(String diagnosticSessionId, String collectionSessionId, String rootTaskId, - String sourceAccount, String warehouseCode, String ownerCode, String orderNo, - Observation listObservation, Observation detailObservation, List skus) { } - public record Observation(String status, String outboundTime, Instant observedAt, String attemptId, - Boolean skuSetComplete) { } - public record Sku(String sku, BigDecimal listQty, BigDecimal forecastQty, BigDecimal actualQty, - Boolean listPresent, Boolean detailPresent) { } - public enum Comparison { - SKU_SET_MISMATCH, QUANTITY_UNKNOWN, FORECAST_ACTUAL_DIFFERENCE, ALL_EQUAL, LIST_QUANTITY_DIFFERENCE - } - public record Receipt(String outcome, Comparison comparison, String stopReason, - int acceptedOrders, int acceptedSkuRows, long acceptedBytes) { } + private QuantityDiagnosticContract() {} + public record Arm(String diagnosticSessionId,String sourceAccount,List warehouseCodes,String ownerCode,String excludedRootTaskId) {} + public record Status(String sourceAccount,String producerId,boolean attach,boolean drained) {} + public record Stop(String diagnosticSessionId,String sourceAccount,String producerId,String reason) {} + public record Observation(String status,String outboundTime,Instant observedAt,String attemptId,Boolean skuSetComplete) {} + public record Sku(String sku,BigDecimal listQty,BigDecimal forecastQty,BigDecimal actualQty) {} + public record Event(String diagnosticSessionId,String sourceAccount,String producerId,String rootTaskId,String collectionSessionId,String operation,String warehouseCode,String ownerCode,String orderNo,String side,Observation observation,List skus) {} + public record Key(String warehouseCode,String ownerCode,String orderNo,boolean paired) {} + public record View(String diagnosticSessionId,String sourceAccount,String rootTaskId,String collectionSessionId,String state,String stopReason,Instant startsAt,Instant expiresAt,Instant bindBefore,List warehouseCodes,String ownerCode,String excludedRootTaskId,int candidates,int pairedOrders,int retainedRows,long retainedBytes,boolean deliveryUncertain,List selected) {} + public record Receipt(String outcome,String comparison,View session) {} } diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticController.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticController.java index e189db304..a0737bee9 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticController.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticController.java @@ -24,20 +24,35 @@ public class QuantityDiagnosticController { this.json = json.copy().enable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES) .enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS); } - @PostMapping(value = "/orders", consumes = "application/json") - public ApiResponse accept( - @RequestHeader(value = "X-Internal-Service-Token", required = false) String token, @RequestBody byte[] body) { + @PostMapping(value="/arm",consumes="application/json") + public ApiResponse arm(@RequestHeader(value="X-Internal-Service-Token",required=false) String token,jakarta.servlet.http.HttpServletRequest request){ + return run(()->service.arm(read(token,request,QuantityDiagnosticContract.Arm.class))); + } + @PostMapping(value="/status",consumes="application/json") + public ApiResponse status(@RequestHeader(value="X-Internal-Service-Token",required=false) String token,jakarta.servlet.http.HttpServletRequest request){ + return run(()->service.status(read(token,request,QuantityDiagnosticContract.Status.class))); + } + @PostMapping(value="/observe",consumes="application/json") + public ApiResponse observe(@RequestHeader(value="X-Internal-Service-Token",required=false) String token,jakarta.servlet.http.HttpServletRequest request){ + return run(()->service.observe(read(token,request,QuantityDiagnosticContract.Event.class))); + } + @PostMapping(value="/stop",consumes="application/json") + public ApiResponse stop(@RequestHeader(value="X-Internal-Service-Token",required=false) String token,jakarta.servlet.http.HttpServletRequest request){ + return run(()->service.stop(read(token,request,QuantityDiagnosticContract.Stop.class))); + } + private T read(String token,jakarta.servlet.http.HttpServletRequest request,Class type){ guard.verify(token); - if (body.length > QuantityDiagnosticPolicy.MAX_REQUEST_BYTES) - throw QuantityDiagnosticPolicy.invalid("DIAGNOSTIC_REQUEST_TOO_LARGE"); - QuantityDiagnosticContract.Order order; - try { order = json.readValue(body, QuantityDiagnosticContract.Order.class); } - catch (IOException e) { throw QuantityDiagnosticPolicy.invalid("DIAGNOSTIC_JSON_INVALID"); } - try { return ApiResponse.success(service.accept(order)); } - catch (DataAccessException | TransactionException e) { - // No SQL, request payload or credentials are returned to the caller. - throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE, "DIAGNOSTIC_STORAGE_UNAVAILABLE"); - } + int limit=QuantityDiagnosticPolicy.MAX_REQUEST_BYTES; + if(request.getContentLengthLong()>limit)throw QuantityDiagnosticPolicy.invalid("DIAGNOSTIC_REQUEST_TOO_LARGE"); + try { + byte[] body=request.getInputStream().readNBytes(limit+1); + if(body.length>limit)throw QuantityDiagnosticPolicy.invalid("DIAGNOSTIC_REQUEST_TOO_LARGE"); + return json.readValue(body,type); + }catch(IOException e){throw QuantityDiagnosticPolicy.invalid("DIAGNOSTIC_JSON_INVALID");} + } + private ApiResponse run(java.util.function.Supplier action){ + try{return ApiResponse.success(action.get());} + catch(DataAccessException|TransactionException e){throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,"DIAGNOSTIC_STORAGE_UNAVAILABLE");} } @ExceptionHandler(ResponseStatusException.class) public ResponseEntity> rejected(ResponseStatusException error) { diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticPolicy.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticPolicy.java index 0c1ddd085..c8ac75384 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticPolicy.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticPolicy.java @@ -1,89 +1,33 @@ package com.baiyue.lastleg.selfoperated.analytics.diagnostic; - import static com.baiyue.lastleg.selfoperated.analytics.diagnostic.QuantityDiagnosticContract.*; import java.math.BigDecimal; -import java.time.Duration; import java.time.Instant; -import java.util.Comparator; -import java.util.HashSet; -import java.util.List; +import java.util.*; import org.springframework.http.HttpStatus; import org.springframework.web.server.ResponseStatusException; - final class QuantityDiagnosticPolicy { - static final int MAX_ORDERS = 100, MAX_ROWS = 2000, MAX_ORDER_ROWS = 200; - static final int MAX_BYTES = 2 * 1024 * 1024, MAX_REQUEST_BYTES = 256 * 1024; - private QuantityDiagnosticPolicy() { } - static ResponseStatusException invalid(String code) { - return new ResponseStatusException(HttpStatus.BAD_REQUEST, code); - } - static void text(String value, int maximum) { - if (value == null || value.isBlank() || value.length() > maximum - || value.codePoints().anyMatch(Character::isISOControl)) throw invalid("DIAGNOSTIC_FIELD_INVALID"); - } - static void settings(QuantityDiagnosticProperties p) { - text(p.diagnosticSessionId(), 128); text(p.collectionSessionId(), 128); - text(p.rootTaskId(), 128); text(p.sourceAccount(), 64); - if (p.startsAt() == null || p.expiresAt() == null || !p.expiresAt().isAfter(p.startsAt()) - || Duration.between(p.startsAt(), p.expiresAt()).compareTo(Duration.ofMinutes(30)) > 0 - || p.warehouseCodes() == null || p.warehouseCodes().isEmpty() || p.warehouseCodes().size() > 20) - throw invalid("DIAGNOSTIC_CONFIGURATION_INVALID"); - p.warehouseCodes().forEach(w -> text(w, 64)); - } - static Order normalize(Order r, QuantityDiagnosticProperties p, Instant now) { - if (r == null) throw invalid("DIAGNOSTIC_ORDER_REQUIRED"); - if (!p.diagnosticSessionId().equals(r.diagnosticSessionId()) - || !p.collectionSessionId().equals(r.collectionSessionId()) || !p.rootTaskId().equals(r.rootTaskId()) - || !p.sourceAccount().equals(r.sourceAccount()) || !p.warehouseCodes().contains(r.warehouseCode())) - throw invalid("DIAGNOSTIC_SCOPE_MISMATCH"); - text(r.ownerCode(), 128); text(r.orderNo(), 128); - observation(r.listObservation(), p, now); observation(r.detailObservation(), p, now); - if (r.skus() == null || r.skus().isEmpty() || r.skus().size() > MAX_ORDER_ROWS) - throw invalid("DIAGNOSTIC_SKU_LIMIT"); - var seen = new HashSet(); - for (Sku s : r.skus()) { - if (s == null) throw invalid("DIAGNOSTIC_SKU_REQUIRED"); - text(s.sku(), 128); - if (!seen.add(s.sku())) throw invalid("DIAGNOSTIC_DUPLICATE_SKU"); - if (s.listPresent() == null || s.detailPresent() == null || (!s.listPresent() && !s.detailPresent())) - throw invalid("DIAGNOSTIC_PRESENCE_INVALID"); - if ((!s.listPresent() && s.listQty() != null) - || (!s.detailPresent() && (s.forecastQty() != null || s.actualQty() != null))) - throw invalid("DIAGNOSTIC_ABSENT_QUANTITY"); - quantity(s.listQty()); quantity(s.forecastQty()); quantity(s.actualQty()); - } - List skus = r.skus().stream().sorted(Comparator.comparing(Sku::sku)) - .map(s -> new Sku(s.sku(), decimal(s.listQty()), decimal(s.forecastQty()), decimal(s.actualQty()), - s.listPresent(), s.detailPresent())).toList(); - return new Order(r.diagnosticSessionId(), r.collectionSessionId(), r.rootTaskId(), r.sourceAccount(), - r.warehouseCode(), r.ownerCode(), r.orderNo(), r.listObservation(), r.detailObservation(), skus); - } - private static void observation(Observation o, QuantityDiagnosticProperties p, Instant now) { - if (o == null || o.observedAt() == null || o.skuSetComplete() == null) - throw invalid("DIAGNOSTIC_OBSERVATION_REQUIRED"); - text(o.attemptId(), 128); - if (o.status() != null) text(o.status(), 32); - if (o.outboundTime() != null) text(o.outboundTime(), 32); - if (o.observedAt().isBefore(p.startsAt()) || o.observedAt().isAfter(p.expiresAt()) - || o.observedAt().isAfter(now)) throw invalid("DIAGNOSTIC_OBSERVATION_OUTSIDE_SESSION"); - } - private static BigDecimal decimal(BigDecimal v) { return v == null ? null : v.stripTrailingZeros(); } - private static void quantity(BigDecimal v) { - if (v != null && (v.signum() < 0 || v.scale() > 6 || v.precision() - v.scale() > 32)) - throw invalid("DIAGNOSTIC_QUANTITY_INVALID"); - } - static boolean forecastDifference(Order r) { - return r.skus().stream().anyMatch(s -> s.forecastQty() != null && s.actualQty() != null - && s.forecastQty().compareTo(s.actualQty()) != 0); - } - static Comparison compare(Order r) { - if (!r.listObservation().skuSetComplete() || !r.detailObservation().skuSetComplete() - || r.skus().stream().anyMatch(s -> !s.listPresent() || !s.detailPresent())) - return Comparison.SKU_SET_MISMATCH; - if (r.skus().stream().anyMatch(s -> s.listQty() == null || s.forecastQty() == null || s.actualQty() == null)) - return Comparison.QUANTITY_UNKNOWN; - if (forecastDifference(r)) return Comparison.FORECAST_ACTUAL_DIFFERENCE; - return r.skus().stream().allMatch(s -> s.listQty().compareTo(s.actualQty()) == 0) - ? Comparison.ALL_EQUAL : Comparison.LIST_QUANTITY_DIFFERENCE; - } + static final int MAX_ORDERS=100,MAX_ROWS=2000,MAX_BYTES=2097152,MAX_REQUEST_BYTES=262144; + private QuantityDiagnosticPolicy() {} + static ResponseStatusException invalid(String code){return new ResponseStatusException(HttpStatus.BAD_REQUEST,code);} + static void require(boolean value,String code){if(!value)throw invalid(code);} + static void text(String s,int max){require(s!=null&&!s.isBlank()&&s.length()<=max&&s.codePoints().noneMatch(Character::isISOControl),"DIAGNOSTIC_FIELD_INVALID");} + static Event normalize(Event e,Instant now){ + require(e!=null,"DIAGNOSTIC_EVENT_REQUIRED");text(e.diagnosticSessionId(),128);text(e.sourceAccount(),64);text(e.rootTaskId(),128);text(e.collectionSessionId(),128);text(e.warehouseCode(),64);text(e.ownerCode(),128);text(e.orderNo(),128); + text(e.producerId(),128);require("LIST".equals(e.side())||"DETAIL".equals(e.side()),"DIAGNOSTIC_SIDE_INVALID");require(("LIST".equals(e.side())?"OUTBOUND_QUERY":"OUTBOUND_DETAIL").equals(e.operation()),"DIAGNOSTIC_OPERATION_INVALID");require(e.observation()!=null&&e.observation().observedAt()!=null&&e.observation().skuSetComplete()!=null,"DIAGNOSTIC_OBSERVATION_REQUIRED");text(e.observation().attemptId(),128); + if(e.observation().status()!=null)text(e.observation().status(),32);if(e.observation().outboundTime()!=null)text(e.observation().outboundTime(),64); + require(!e.observation().observedAt().isAfter(now),"DIAGNOSTIC_FUTURE_OBSERVATION");require(e.skus()!=null&&(!e.skus().isEmpty()||!e.observation().skuSetComplete())&&e.skus().size()<=200,"DIAGNOSTIC_SKU_LIMIT"); + var seen=new HashSet();var skus=new ArrayList();for(var s:e.skus()){require(s!=null,"DIAGNOSTIC_SKU_REQUIRED");text(s.sku(),128);require(seen.add(s.sku()),"DIAGNOSTIC_DUPLICATE_SKU"); + for(var n:Arrays.asList(s.listQty(),s.forecastQty(),s.actualQty()))if(n!=null)require(n.signum()>=0&&n.scale()<=6&&n.precision()-n.scale()<=32,"DIAGNOSTIC_QUANTITY_INVALID"); + require("LIST".equals(e.side())?s.forecastQty()==null&&s.actualQty()==null:s.listQty()==null,"DIAGNOSTIC_SIDE_FIELDS");skus.add(new Sku(s.sku(),decimal(s.listQty()),decimal(s.forecastQty()),decimal(s.actualQty()))); + }skus.sort(Comparator.comparing(Sku::sku));return new Event(e.diagnosticSessionId(),e.sourceAccount(),e.producerId(),e.rootTaskId(),e.collectionSessionId(),e.operation(),e.warehouseCode(),e.ownerCode(),e.orderNo(),e.side(),e.observation(),List.copyOf(skus)); + } + static BigDecimal decimal(BigDecimal n){return n==null?null:n.stripTrailingZeros();} + static boolean different(BigDecimal a,BigDecimal b){return a!=null&&b!=null&&a.compareTo(b)!=0;} + static String compare(Event list,Event detail){ + var l=new TreeMap();var d=new TreeMap();list.skus().forEach(s->l.put(s.sku(),s));detail.skus().forEach(s->d.put(s.sku(),s)); + boolean incomplete=!Boolean.TRUE.equals(list.observation().skuSetComplete())||!Boolean.TRUE.equals(detail.observation().skuSetComplete())||!l.keySet().equals(d.keySet()); + var keys=new HashSet(l.keySet());keys.addAll(d.keySet());boolean difference=false; + for(String key:keys){var a=l.get(key);var b=d.get(key);BigDecimal x=a==null?null:a.listQty(),y=b==null?null:b.forecastQty(),z=b==null?null:b.actualQty();difference|=different(x,y)||different(x,z)||different(y,z);incomplete|=x==null||y==null||z==null;} + return incomplete?"INCONCLUSIVE":difference?"QUANTITY_DIFFERENCE":"ALL_EQUAL"; + } } diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticProperties.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticProperties.java index d65207e9e..630aa6d39 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticProperties.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticProperties.java @@ -1,11 +1,6 @@ package com.baiyue.lastleg.selfoperated.analytics.diagnostic; - -import java.time.Instant; import java.util.List; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.boot.context.properties.bind.DefaultValue; - @ConfigurationProperties("baiyue.selfop.quantity-diagnostics") -public record QuantityDiagnosticProperties(@DefaultValue("false") boolean enabled, - String diagnosticSessionId, String collectionSessionId, String rootTaskId, String sourceAccount, - List warehouseCodes, Instant startsAt, Instant expiresAt) { } +public record QuantityDiagnosticProperties(@DefaultValue("false") boolean enabled, String sourceAccount, List warehouseCodes) {} diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticService.java index c74396862..995d78962 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticService.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticService.java @@ -2,124 +2,218 @@ package com.baiyue.lastleg.selfoperated.analytics.diagnostic; import static com.baiyue.lastleg.selfoperated.analytics.diagnostic.QuantityDiagnosticContract.*; import static com.baiyue.lastleg.selfoperated.analytics.diagnostic.QuantityDiagnosticPolicy.*; -import com.fasterxml.jackson.core.JsonProcessingException; +import com.baiyue.lastleg.selfoperated.analytics.model.SelfopSyncTask; +import com.baiyue.lastleg.selfoperated.analytics.model.SelfopCollectionPolicy; import com.fasterxml.jackson.databind.ObjectMapper; +import com.zaxxer.hikari.HikariDataSource; +import jakarta.annotation.PreDestroy; +import javax.sql.DataSource; import java.nio.charset.StandardCharsets; import java.security.MessageDigest; -import java.security.NoSuchAlgorithmException; -import java.time.Clock; -import java.time.Instant; -import java.util.HexFormat; -import java.util.List; +import java.time.*; +import java.util.*; +import java.util.concurrent.Semaphore; +import java.util.function.Supplier; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.http.HttpStatus; import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.core.BeanPropertyRowMapper; +import org.springframework.jdbc.datasource.DataSourceTransactionManager; import org.springframework.stereotype.Service; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.support.TransactionTemplate; import org.springframework.web.server.ResponseStatusException; -/** One short transaction per paired order; only the two diagnostic tables are written. */ +/** Serializes diagnostic state only; never locks or updates collection/business rows. */ @Service public class QuantityDiagnosticService { private final JdbcTemplate jdbc; private final TransactionTemplate tx; private final ObjectMapper json; - private final QuantityDiagnosticProperties properties; + private final QuantityDiagnosticProperties config; private final Clock clock; + private final Semaphore admission = new Semaphore(1); + private HikariDataSource pool; @Autowired - public QuantityDiagnosticService(JdbcTemplate jdbc, PlatformTransactionManager manager, - ObjectMapper json, QuantityDiagnosticProperties properties) { - this(jdbc, manager, json, properties, Clock.systemUTC()); - } - QuantityDiagnosticService(JdbcTemplate jdbc, PlatformTransactionManager manager, - ObjectMapper json, QuantityDiagnosticProperties properties, Clock clock) { - this.jdbc = jdbc; this.json = json.copy(); this.properties = properties; this.clock = clock; - this.tx = new TransactionTemplate(manager); this.tx.setTimeout(3); - } - public Receipt accept(Order input) { - if (!properties.enabled()) return new Receipt("STOPPED", null, "DISABLED", 0, 0, 0); - settings(properties); - Order order = normalize(input, properties, clock.instant()); - String payload = encode(order); - int bytes = payload.getBytes(StandardCharsets.UTF_8).length; - if (bytes > MAX_REQUEST_BYTES) throw invalid("DIAGNOSTIC_REQUEST_TOO_LARGE"); - String key = hash(encode(List.of(order.sourceAccount(), order.warehouseCode(), order.ownerCode(), order.orderNo()))); - String payloadHash = hash(payload); - String configuration = hash(encode(List.of("v1", properties.diagnosticSessionId(), properties.collectionSessionId(), - properties.rootTaskId(), properties.sourceAccount(), properties.warehouseCodes().stream().sorted().distinct().toList(), - properties.startsAt(), properties.expiresAt(), MAX_ORDERS, MAX_ROWS, MAX_BYTES))); - return tx.execute(status -> persist(order, key, payload, payloadHash, bytes, configuration)); - } - private Receipt persist(Order order, String key, String payload, String payloadHash, int bytes, String configuration) { - String id = order.diagnosticSessionId(); - // Duplicate insert acquires the same durable session lock, including first-request races. - jdbc.update(""" - INSERT INTO ll_selfop_quantity_diagnostic_session - (diagnostic_session_id, configuration_hash, source_account, collection_session_id, root_task_id, - starts_at, expires_at, accepted_orders, accepted_sku_rows, accepted_bytes) - VALUES (?, ?, ?, ?, ?, ?, ?, 0, 0, 0) - ON DUPLICATE KEY UPDATE diagnostic_session_id = diagnostic_session_id - """, id, configuration, order.sourceAccount(), order.collectionSessionId(), order.rootTaskId(), - properties.startsAt().toString(), properties.expiresAt().toString()); - Session session = jdbc.queryForObject(""" - SELECT configuration_hash, accepted_orders, accepted_sku_rows, accepted_bytes, stop_reason - FROM ll_selfop_quantity_diagnostic_session WHERE diagnostic_session_id = ? FOR UPDATE - """, (rs, n) -> new Session(rs.getString(1), rs.getInt(2), rs.getInt(3), rs.getLong(4), rs.getString(5)), id); - if (session == null || !configuration.equals(session.configurationHash())) - throw new ResponseStatusException(HttpStatus.CONFLICT, "DIAGNOSTIC_SESSION_CONFIGURATION_CHANGED"); - var existing = jdbc.query(""" - SELECT payload_hash, comparison FROM ll_selfop_quantity_diagnostic_order - WHERE diagnostic_session_id = ? AND order_key_hash = ? - """, (rs, n) -> new Existing(rs.getString(1), Comparison.valueOf(rs.getString(2))), id, key); - if (!existing.isEmpty()) { - if (!payloadHash.equals(existing.get(0).payloadHash())) - throw new ResponseStatusException(HttpStatus.CONFLICT, "DIAGNOSTIC_ORDER_CONFLICT"); - return session.receipt("DUPLICATE", existing.get(0).comparison(), session.stopReason()); + public QuantityDiagnosticService(DataSource source, ObjectMapper json, QuantityDiagnosticProperties config) { + this.json=json.copy().disable(com.fasterxml.jackson.databind.SerializationFeature.WRITE_DATES_AS_TIMESTAMPS); + this.config=config; this.clock=Clock.systemUTC(); + // Separate, bounded connection acquisition. Diagnostics cannot exhaust the collection pool. + if(config.enabled()) { + try { + HikariDataSource base=source.unwrap(HikariDataSource.class); + pool=new HikariDataSource(); pool.setJdbcUrl(base.getJdbcUrl()); pool.setUsername(base.getUsername()); + pool.setPassword(base.getPassword()); pool.setDataSourceProperties(base.getDataSourceProperties()); + pool.setMaximumPoolSize(1); pool.setMinimumIdle(0); pool.setConnectionTimeout(250); + pool.setValidationTimeout(250); pool.setInitializationFailTimeout(-1); pool.setPoolName("quantity-diagnostic"); + } catch(Exception unavailable) { if(pool!=null)pool.close(); pool=null; } } - if (session.stopReason() != null) return session.receipt("STOPPED", null, session.stopReason()); - Instant now = clock.instant(); - if (now.isBefore(properties.startsAt())) return session.receipt("STOPPED", null, "NOT_STARTED"); - if (!now.isBefore(properties.expiresAt())) return stop(id, session, "EXPIRED"); - if (session.orders() >= MAX_ORDERS || session.rows() + order.skus().size() > MAX_ROWS - || session.bytes() + bytes > MAX_BYTES) return stop(id, session, "LIMIT_REACHED"); - Comparison comparison = compare(order); - jdbc.update(""" - INSERT INTO ll_selfop_quantity_diagnostic_order - (diagnostic_session_id, order_key_hash, source_account, warehouse_code, owner_code, order_no, - payload_hash, comparison, sku_rows, payload_bytes, evidence_json, received_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - """, id, key, order.sourceAccount(), order.warehouseCode(), order.ownerCode(), order.orderNo(), - payloadHash, comparison.name(), order.skus().size(), bytes, payload, now.toString()); - int orders = session.orders() + 1, rows = session.rows() + order.skus().size(); - long totalBytes = session.bytes() + bytes; - String reason = forecastDifference(order) ? "FORECAST_ACTUAL_DIFFERENCE" - : orders == MAX_ORDERS || rows == MAX_ROWS || totalBytes == MAX_BYTES ? "LIMIT_REACHED" : null; - int updated = jdbc.update(""" - UPDATE ll_selfop_quantity_diagnostic_session - SET accepted_orders = ?, accepted_sku_rows = ?, accepted_bytes = ?, stop_reason = ? - WHERE diagnostic_session_id = ? - """, orders, rows, totalBytes, reason, id); - if (updated != 1) throw new IllegalStateException("DIAGNOSTIC_SESSION_UPDATE_FAILED"); - return new Receipt("ACCEPTED", comparison, reason, orders, rows, totalBytes); + jdbc=pool==null?null:new JdbcTemplate(pool); + tx=pool==null?null:new TransactionTemplate(new DataSourceTransactionManager(pool)); + if(tx!=null) { tx.setTimeout(3); jdbc.setQueryTimeout(2); } } - private Receipt stop(String id, Session session, String reason) { - jdbc.update("UPDATE ll_selfop_quantity_diagnostic_session SET stop_reason = ? WHERE diagnostic_session_id = ?", reason, id); - return session.receipt("STOPPED", null, reason); + QuantityDiagnosticService(JdbcTemplate jdbc,PlatformTransactionManager manager,ObjectMapper json, + QuantityDiagnosticProperties config,Clock clock) { + this.jdbc=jdbc;this.json=json.copy().disable(com.fasterxml.jackson.databind.SerializationFeature.WRITE_DATES_AS_TIMESTAMPS); + this.config=config;this.clock=clock;tx=new TransactionTemplate(manager);tx.setTimeout(3); } - private String encode(Object value) { - try { return json.writeValueAsString(value); } - catch (JsonProcessingException e) { throw invalid("DIAGNOSTIC_ENCODING_FAILED"); } + @PreDestroy public void close(){if(pool!=null)pool.close();} + private T guarded(Supplier action){ + require(config.enabled(),"DIAGNOSTIC_DISABLED"); + if(tx==null)throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,"DIAGNOSTIC_DATASOURCE_UNAVAILABLE"); + if(!admission.tryAcquire())throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,"DIAGNOSTIC_BUSY"); + try{return tx.execute(s->action.get());}finally{admission.release();} } - private static String hash(String value) { - try { return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(value.getBytes(StandardCharsets.UTF_8))); } - catch (NoSuchAlgorithmException e) { throw new IllegalStateException(e); } + private void account(String account){text(account,64);require(account.equals(config.sourceAccount()),"DIAGNOSTIC_ACCOUNT_REJECTED");} + public View arm(Arm request){ + require(request!=null,"DIAGNOSTIC_ARM_REQUIRED");text(request.diagnosticSessionId(),128);account(request.sourceAccount()); + text(request.ownerCode(),128);if(request.excludedRootTaskId()!=null)text(request.excludedRootTaskId(),128); + require(request.warehouseCodes()!=null&&!request.warehouseCodes().isEmpty()&&request.warehouseCodes().size()<=8,"DIAGNOSTIC_SCOPE_REJECTED"); + request.warehouseCodes().forEach(w->text(w,64)); + require(config.warehouseCodes()!=null&&config.warehouseCodes().containsAll(request.warehouseCodes()),"DIAGNOSTIC_SCOPE_REJECTED"); + Arm scope=new Arm(request.diagnosticSessionId(),request.sourceAccount(),request.warehouseCodes().stream().sorted().distinct().toList(),request.ownerCode(),request.excludedRootTaskId()); + return guarded(()->{ + Session existing=load(scope.diagnosticSessionId()); + if(existing!=null){require(existing.scope.equals(scope),"DIAGNOSTIC_ARM_CONFLICT");advance(existing);save(existing);return view(existing);} + // Read-only arm snapshot: unfinished automatic roots that have actually started. + var current=jdbc.queryForList("SELECT id FROM ll_selfop_sync_task WHERE source_account=? AND task_role='ACCOUNT_DISCOVERY' AND task_type='BACKFILL' AND execution_purpose='BUSINESS' AND requested_by='SELFOP_CONTINUOUS_PLANNER' AND (status IN ('RUNNING','RETRY_WAIT') OR (status='READY' AND attempt_count>0)) ORDER BY id LIMIT 2",String.class,scope.sourceAccount()); + require(current.size()<=1,"DIAGNOSTIC_CURRENT_ROOT_AMBIGUOUS"); + String excluded=current.isEmpty()?null:current.get(0); + require(Objects.equals(excluded,scope.excludedRootTaskId()),"DIAGNOSTIC_CURRENT_ROOT_CHANGED"); + require(excluded==null||eligible(root(excluded),scope.sourceAccount()),"DIAGNOSTIC_EXCLUDED_ROOT_INVALID"); + var active=jdbc.queryForList("SELECT diagnostic_session_id FROM ll_selfop_quantity_diagnostic_session WHERE source_account=? AND active_slot=1",String.class,scope.sourceAccount()); + if(!active.isEmpty()){Session s=load(active.get(0));advance(s);save(s);require(terminal(s),"DIAGNOSTIC_ALREADY_ARMED");} + Session s=new Session();s.scope=scope;s.armedAt=clock.instant();s.bindBefore=s.armedAt.plusSeconds(21600);s.expiresAt=s.armedAt.plusSeconds(36000);s.state="ARMED"; + // Account+active_slot unique constraint is the cross-instance arm arbitration point. + jdbc.update("INSERT INTO ll_selfop_quantity_diagnostic_session(diagnostic_session_id,configuration_hash,source_account,starts_at,expires_at,active_slot,state_json) VALUES(?,?,?,?,?,1,?)",scope.diagnosticSessionId(),hash(encode(scope)),scope.sourceAccount(),s.armedAt.toString(),s.expiresAt.toString(),encode(s)); + return view(s); + }); } - private record Existing(String payloadHash, Comparison comparison) { } - private record Session(String configurationHash, int orders, int rows, long bytes, String stopReason) { - Receipt receipt(String outcome, Comparison comparison, String reason) { - return new Receipt(outcome, comparison, reason, orders, rows, bytes); - } + public View status(Status request){ + require(request!=null,"DIAGNOSTIC_STATUS_REQUIRED");account(request.sourceAccount());text(request.producerId(),128); + return guarded(()->{ + var ids=jdbc.queryForList("SELECT diagnostic_session_id FROM ll_selfop_quantity_diagnostic_session WHERE source_account=? AND state_json IS NOT NULL ORDER BY active_slot DESC, starts_at DESC LIMIT 1",String.class,request.sourceAccount()); + if(ids.isEmpty())return null; + Session s=load(ids.get(0));advance(s); + if(!terminal(s)){ + if(s.producerId==null)s.producerId=request.producerId(); + else if(!s.producerId.equals(request.producerId())){ + require(request.attach(),"DIAGNOSTIC_PRODUCER_FENCED"); + // Recovery retains persisted candidates, but cannot claim the lost in-memory tail was complete. + s.deliveryUncertain=true;s.producerId=request.producerId(); + } + if("DRAINING".equals(s.state)&&request.drained())finish(s); + } + save(s);return view(s); + }); + } + public View stop(Stop request){ + require(request!=null,"DIAGNOSTIC_STOP_REQUIRED");account(request.sourceAccount());text(request.diagnosticSessionId(),128);text(request.producerId(),128); + require(Set.of("INCOMPLETE","QUEUE_FULL","OBSERVER_FAILED","SHUTDOWN").contains(request.reason()),"DIAGNOSTIC_STOP_REASON"); + return guarded(()->{Session s=load(request.diagnosticSessionId());require(s!=null&&s.scope.sourceAccount().equals(request.sourceAccount()),"DIAGNOSTIC_SESSION_UNKNOWN"); + require(Objects.equals(s.producerId,request.producerId()),"DIAGNOSTIC_PRODUCER_FENCED"); + if(!terminal(s)){s.deliveryUncertain=true;end(s,"INCOMPLETE");}save(s);return view(s);}); + } + public Receipt observe(Event input){ + Event e=normalize(input,clock.instant());account(e.sourceAccount());String payload=encode(e);int bytes=size(payload); + require(bytes<=MAX_REQUEST_BYTES,"DIAGNOSTIC_REQUEST_TOO_LARGE"); + return guarded(()->{ + Session s=load(e.diagnosticSessionId());require(s!=null&&s.scope.sourceAccount().equals(e.sourceAccount()),"DIAGNOSTIC_SESSION_UNKNOWN"); + String key=hash(encode(List.of(e.sourceAccount(),e.warehouseCode(),e.ownerCode(),e.orderNo()))); + var completed=jdbc.query("SELECT evidence_json,comparison FROM ll_selfop_quantity_diagnostic_order WHERE diagnostic_session_id=? AND order_key_hash=?",(r,n)->new String[]{r.getString(1),r.getString(2)},e.diagnosticSessionId(),key); + if(!completed.isEmpty()){ + Pair pair=decode(completed.get(0)[0],Pair.class);Event original="LIST".equals(e.side())?pair.list:pair.detail; + require(encode(original).equals(payload),"DIAGNOSTIC_OBSERVATION_CONFLICT"); + return new Receipt("DUPLICATE",completed.get(0)[1],view(s)); + } + var pending=jdbc.query("SELECT evidence_json FROM ll_selfop_quantity_diagnostic_pending WHERE diagnostic_session_id=? AND order_key_hash=?",(r,n)->decode(r.getString(1),Event.class),e.diagnosticSessionId(),key); + if("LIST".equals(e.side())&&!pending.isEmpty()){ + require(encode(pending.get(0)).equals(payload),"DIAGNOSTIC_OBSERVATION_CONFLICT");return new Receipt("DUPLICATE",null,view(s)); + } + // Session lock is the receiver acceptance boundary. Integration enqueue is not acceptance. + advance(s); + if(terminal(s)){save(s);return new Receipt("STOPPED",null,view(s));} + require(Objects.equals(s.producerId,e.producerId()),"DIAGNOSTIC_PRODUCER_FENCED"); + require(!e.observation().observedAt().isBefore(s.armedAt),"DIAGNOSTIC_OLD_OBSERVATION"); + require(s.scope.warehouseCodes().contains(e.warehouseCode())&&("*".equals(s.scope.ownerCode())||s.scope.ownerCode().equals(e.ownerCode())),"DIAGNOSTIC_SCOPE_REJECTED"); + require(("selfop-"+e.rootTaskId()).equals(e.collectionSessionId()),"DIAGNOSTIC_COLLECTION_SESSION_REJECTED"); + SelfopSyncTask task=root(e.rootTaskId());require(eligible(task,e.sourceAccount()),"DIAGNOSTIC_ROOT_REJECTED"); + if(s.rootTaskId==null){ + if(!"LIST".equals(e.side())||e.rootTaskId().equals(s.scope.excludedRootTaskId())||!clock.instant().isBefore(s.bindBefore)||!e.observation().observedAt().isBefore(s.bindBefore))return new Receipt("IGNORED",null,view(s)); + s.rootTaskId=e.rootTaskId();s.collectionSessionId=e.collectionSessionId();s.boundAt=e.observation().observedAt(); + s.expiresAt=s.boundAt.plusSeconds(14400).isBefore(s.expiresAt)?s.boundAt.plusSeconds(14400):s.expiresAt;s.state="BOUND"; + advance(s); + if(terminal(s)){save(s);return new Receipt("STOPPED",null,view(s));} + } + require(s.rootTaskId.equals(e.rootTaskId())&&s.collectionSessionId.equals(e.collectionSessionId()),"DIAGNOSTIC_ROOT_FROZEN"); + if(ended(task)&&!"DRAINING".equals(s.state)){s.state="DRAINING";s.rootEndedAt=clock.instant();} + if("LIST".equals(e.side())&&("DRAINING".equals(s.state)||s.candidates>=MAX_ORDERS)){save(s);return new Receipt("IGNORED",null,view(s));} + if("DETAIL".equals(e.side())&&pending.isEmpty()){save(s);return new Receipt("IGNORED",null,view(s));} + if(s.rootEndedAt!=null&&e.observation().observedAt().isAfter(s.rootEndedAt)){save(s);return new Receipt("IGNORED",null,view(s));} + // Retained bytes/rows are monotonic and count both observations; promotion never refunds quota. + int overhead="LIST".equals(e.side())?20:0; + if(s.rows+e.skus().size()>MAX_ROWS||s.bytes+bytes+overhead>MAX_BYTES){end(s,"INCOMPLETE");save(s);return new Receipt("STOPPED",null,view(s));} + s.rows+=e.skus().size();s.bytes+=bytes+overhead; + String comparison=null; + if("LIST".equals(e.side())){ + s.candidates++; + jdbc.update("INSERT INTO ll_selfop_quantity_diagnostic_pending(diagnostic_session_id,order_key_hash,warehouse_code,owner_code,order_no,evidence_json) VALUES(?,?,?,?,?,?)",e.diagnosticSessionId(),key,e.warehouseCode(),e.ownerCode(),e.orderNo(),payload); + if(!Boolean.TRUE.equals(e.observation().skuSetComplete())||e.skus().stream().anyMatch(v->v.listQty()==null)){comparison="INCONCLUSIVE";end(s,comparison);} + }else{ + Event list=pending.get(0);comparison=compare(list,e);String evidence=encode(new Pair(list,e));s.paired++; + require(size(evidence)<=MAX_REQUEST_BYTES,"DIAGNOSTIC_PAIR_TOO_LARGE"); + jdbc.update("INSERT INTO ll_selfop_quantity_diagnostic_order(diagnostic_session_id,order_key_hash,source_account,warehouse_code,owner_code,order_no,payload_hash,comparison,sku_rows,payload_bytes,evidence_json,received_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)",e.diagnosticSessionId(),key,e.sourceAccount(),e.warehouseCode(),e.ownerCode(),e.orderNo(),hash(evidence),comparison,Math.max(list.skus().size(),e.skus().size()),size(evidence),evidence,clock.instant().toString()); + jdbc.update("DELETE FROM ll_selfop_quantity_diagnostic_pending WHERE diagnostic_session_id=? AND order_key_hash=?",e.diagnosticSessionId(),key); + if(!"ALL_EQUAL".equals(comparison))end(s,comparison); + else if(s.paired==MAX_ORDERS)end(s,s.deliveryUncertain?"INCOMPLETE":"COMPLETED"); + } + save(s);return new Receipt("ACCEPTED",comparison,view(s)); + }); + } + private void advance(Session s){ + if(terminal(s))return; + Instant now=clock.instant(); + if(!now.isBefore(s.expiresAt)){end(s,"INCOMPLETE");return;} + if(s.rootTaskId==null){if(!now.isBefore(s.bindBefore))end(s,"BIND_TIMEOUT");return;} + if(ended(root(s.rootTaskId))&&!"DRAINING".equals(s.state)){s.state="DRAINING";s.rootEndedAt=now;} + } + private void finish(Session s){end(s,s.deliveryUncertain||s.candidates!=s.paired?"INCOMPLETE":s.paired(SelfopSyncTask.class),id); + return rows.isEmpty()?null:rows.get(0); + } + private Session load(String id){ + var sessions=jdbc.query("SELECT state_json,root_task_id,collection_session_id,starts_at,expires_at,accepted_orders,accepted_sku_rows,accepted_bytes,stop_reason FROM ll_selfop_quantity_diagnostic_session WHERE diagnostic_session_id=? FOR UPDATE",(r,n)->{ + String metadata=r.getString(1);require(metadata!=null,"DIAGNOSTIC_HISTORICAL_ID_READ_ONLY");Session s=decode(metadata,Session.class); + // Existing columns remain the sole authority for counters, root, deadline and stop reason. + s.rootTaskId=r.getString(2);s.collectionSessionId=r.getString(3);s.boundAt=s.rootTaskId==null?null:Instant.parse(r.getString(4));s.expiresAt=Instant.parse(r.getString(5)); + s.paired=r.getInt(6);s.rows=r.getInt(7);s.bytes=r.getLong(8);s.reason=r.getString(9); + s.state=s.reason!=null?"TERMINAL":s.rootTaskId==null?"ARMED":s.rootEndedAt!=null?"DRAINING":"BOUND";return s; + },id);return sessions.isEmpty()?null:sessions.get(0); + } + private void save(Session s){jdbc.update("UPDATE ll_selfop_quantity_diagnostic_session SET state_json=?,active_slot=?,collection_session_id=?,root_task_id=?,starts_at=?,expires_at=?,accepted_orders=?,accepted_sku_rows=?,accepted_bytes=?,stop_reason=? WHERE diagnostic_session_id=?",encode(s),terminal(s)?null:1,s.collectionSessionId,s.rootTaskId,(s.boundAt==null?s.armedAt:s.boundAt).toString(),s.expiresAt.toString(),s.paired,s.rows,s.bytes,s.reason,s.scope.diagnosticSessionId());} + private View view(Session s){ + var selected=jdbc.query("SELECT warehouse_code,owner_code,order_no,0 AS paired FROM ll_selfop_quantity_diagnostic_pending WHERE diagnostic_session_id=? UNION ALL SELECT warehouse_code,owner_code,order_no,1 AS paired FROM ll_selfop_quantity_diagnostic_order WHERE diagnostic_session_id=?",(r,n)->new Key(r.getString(1),r.getString(2),r.getString(3),r.getBoolean(4)),s.scope.diagnosticSessionId(),s.scope.diagnosticSessionId()); + return new View(s.scope.diagnosticSessionId(),s.scope.sourceAccount(),s.rootTaskId,s.collectionSessionId,s.state,s.reason,s.boundAt==null?s.armedAt:s.boundAt,s.expiresAt,s.bindBefore,s.scope.warehouseCodes(),s.scope.ownerCode(),s.scope.excludedRootTaskId(),s.candidates,s.paired,s.rows,s.bytes,s.deliveryUncertain,selected); + } + private String encode(Object value){try{return json.writeValueAsString(value);}catch(Exception e){throw invalid("DIAGNOSTIC_ENCODING_FAILED");}} + private T decode(String value,Class type){try{return json.readValue(value,type);}catch(Exception e){throw new IllegalStateException("DIAGNOSTIC_STORAGE_INVALID");}} + private static int size(String s){return s.getBytes(StandardCharsets.UTF_8).length;} + private static String hash(String s){try{return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(s.getBytes(StandardCharsets.UTF_8)));}catch(Exception e){throw new IllegalStateException(e);}} + public record Pair(Event list,Event detail){} + public static class Session { + public Arm scope;public String producerId; + public Instant armedAt,bindBefore,rootEndedAt; + public int candidates;public boolean deliveryUncertain; + @com.fasterxml.jackson.annotation.JsonIgnore public String state,reason,rootTaskId,collectionSessionId; + @com.fasterxml.jackson.annotation.JsonIgnore public Instant boundAt,expiresAt; + @com.fasterxml.jackson.annotation.JsonIgnore public int paired,rows; + @com.fasterxml.jackson.annotation.JsonIgnore public long bytes; } } diff --git a/lastleg-service/src/main/resources/db/selfop-quantity-diagnostics.sql b/lastleg-service/src/main/resources/db/selfop-quantity-diagnostics.sql index eec9e8ae3..deaac3f6a 100644 --- a/lastleg-service/src/main/resources/db/selfop-quantity-diagnostics.sql +++ b/lastleg-service/src/main/resources/db/selfop-quantity-diagnostics.sql @@ -2,16 +2,20 @@ -- Run only after explicit TEST SQL approval. Not executed by service startup. CREATE TABLE ll_selfop_quantity_diagnostic_session ( diagnostic_session_id VARCHAR(128) NOT NULL COMMENT '固定有界数量诊断会话标识', - configuration_hash CHAR(64) NOT NULL COMMENT '冻结范围、UTC起止时间及限额的SHA256摘要', + configuration_hash CHAR(64) NOT NULL COMMENT '冻结授权范围的SHA256摘要', source_account VARCHAR(64) NOT NULL COMMENT '来源账号别名，不保存凭据', - collection_session_id VARCHAR(128) NOT NULL COMMENT '被观察的来源采集会话标识', - root_task_id VARCHAR(128) NOT NULL COMMENT '被观察的账号采集根任务标识', + collection_session_id VARCHAR(128) NULL COMMENT '被观察的来源采集会话标识', + root_task_id VARCHAR(128) NULL COMMENT '被观察的账号采集根任务标识', starts_at VARCHAR(40) NOT NULL COMMENT '诊断开始时刻，ISO8601 UTC字符串', - expires_at VARCHAR(40) NOT NULL COMMENT '诊断截止时刻，距开始最多30分钟，ISO8601 UTC字符串', + expires_at VARCHAR(40) NOT NULL COMMENT '诊断截止时刻，首个LIST观测后最多4小时且授权后最多10小时，ISO8601 UTC字符串', accepted_orders INT NOT NULL DEFAULT 0 COMMENT '已接收完整配对订单数，跨重启累计', - accepted_sku_rows INT NOT NULL DEFAULT 0 COMMENT '已接收归并SKU行数，跨重启累计', + accepted_sku_rows INT NOT NULL DEFAULT 0 COMMENT '两侧已保留SKU行数，跨重启累计', accepted_bytes BIGINT NOT NULL DEFAULT 0 COMMENT '已持久化白名单规范JSON的UTF8字节数', stop_reason VARCHAR(64) NULL COMMENT '诊断终止原因，空值表示尚未终止', + active_slot INT NULL COMMENT '当前授权占位，终态释放；旧证据不自动授权', + state_json LONGTEXT NULL COMMENT '新授权的持久生命周期；NULL仅表示未参与此协议的历史证据', + UNIQUE KEY uq_qdiag_active (source_account, active_slot), + CONSTRAINT chk_qdiag_active CHECK (active_slot IS NULL OR active_slot = 1), PRIMARY KEY (diagnostic_session_id), CONSTRAINT chk_selfop_qdiag_limits CHECK (accepted_orders BETWEEN 0 AND 100 AND accepted_sku_rows BETWEEN 0 AND 2000 AND accepted_bytes BETWEEN 0 AND 2097152) @@ -26,7 +30,7 @@ CREATE TABLE ll_selfop_quantity_diagnostic_order ( order_no VARCHAR(128) NOT NULL COMMENT '来源出库单号', payload_hash CHAR(64) NOT NULL COMMENT '白名单规范载荷SHA256摘要，用于幂等校验', comparison VARCHAR(40) NOT NULL COMMENT 'SKU集合与三类数量比较结果', - sku_rows INT NOT NULL COMMENT '本单归并后的SKU并集行数', + sku_rows INT NOT NULL COMMENT '本单两侧较大的SKU行数；两侧累计额度由会话约束', payload_bytes INT NOT NULL COMMENT '本单白名单规范JSON的UTF8字节数', evidence_json JSON NOT NULL COMMENT '类型化列表与详情观测及SKU三量证据，不含原始响应或收件信息', received_at VARCHAR(40) NOT NULL COMMENT '接收时刻，ISO8601 UTC字符串', @@ -36,3 +40,14 @@ CREATE TABLE ll_selfop_quantity_diagnostic_order ( CONSTRAINT chk_selfop_qdiag_order_limits CHECK (sku_rows BETWEEN 1 AND 200 AND payload_bytes BETWEEN 1 AND 262144) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT='独立配对订单数量诊断证据，不参与销量发布'; + +CREATE TABLE ll_selfop_quantity_diagnostic_pending ( + diagnostic_session_id VARCHAR(128) NOT NULL, + order_key_hash CHAR(64) NOT NULL, + warehouse_code VARCHAR(64) NOT NULL, + owner_code VARCHAR(128) NOT NULL, + order_no VARCHAR(128) NOT NULL, + evidence_json LONGTEXT NOT NULL, + PRIMARY KEY(diagnostic_session_id,order_key_hash), + FOREIGN KEY(diagnostic_session_id) REFERENCES ll_selfop_quantity_diagnostic_session(diagnostic_session_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT='首次LIST白名单证据；配对后同事务移至最终证据表'; diff --git a/lastleg-service/src/main/resources/openapi/selfop-quantity-diagnostics-internal-v1.yaml b/lastleg-service/src/main/resources/openapi/selfop-quantity-diagnostics-internal-v1.yaml index 63708501c..82391d797 100644 --- a/lastleg-service/src/main/resources/openapi/selfop-quantity-diagnostics-internal-v1.yaml +++ b/lastleg-service/src/main/resources/openapi/selfop-quantity-diagnostics-internal-v1.yaml @@ -1,40 +1,168 @@ openapi: 3.0.3 info: - title: Internal bounded self-operated quantity diagnostics - version: 1.0.0 + title: Internal next-root bounded quantity diagnostics + version: 2.0.0 + description: Explicit arm only; no automatic rearm. Bind next normal automatic historical OUTBOUND_QUERY root + within 6h. Sampling expires 4h after first LIST observedAt, never later than armedAt+10h. Lastleg session lock + is receiver acceptance. Source root creation may predate arm. LOCATE, manual/validation roots, current excluded + root and different bound root are rejected/ignored. At most 100 distinct LIST candidates total, 2000 retained + SKU rows across both sides, 2MiB canonical evidence plus promotion overhead. Pending-to-final never refunds + quota. First unknown or SKU mismatch persists INCONCLUSIVE and stops; any known inequality among list/forecast/actual + persists QUANTITY_DIFFERENCE and stops. Original side timestamps/attempts are immutable; exact replay returns + stored receipt even after termination, changed same-side evidence is rejected. Root termination drains only + accepted candidates via an explicit Integration FIFO boundary without extending expiry. Fewer than 100 complete + pairs is ROOT_ENDED_INSUFFICIENT; unpaired/lost tail is INCOMPLETE. Producer restart retains persisted candidates + but marks delivery uncertain. Every request/response is stream-bounded to 262144 bytes. Never exposed through + Gateway, never creates supplier requests or changes business facts. paths: - /internal/lastleg/self-operated-analytics/quantity-diagnostics/orders: + /internal/lastleg/self-operated-analytics/quantity-diagnostics/arm: post: - operationId: receiveSelfopQuantityDiagnosticOrder - description: >- - Internal service-token only; never routed through Gateway. One complete paired order per request. - Quantities are evidence only. Session scope and UTC time interval are frozen by server configuration. - Maximum 100 orders, 2000 SKU rows and 2097152 canonical UTF8 bytes per session; - 200 SKU rows and 262144 wire/canonical bytes per request; at most 30 minutes. - First known forecast/actual difference closes capture regardless of comparison classification. - Same identity and canonical evidence is idempotent, including after closure; changed evidence is 409. + operationId: quantityDiagnosticArm security: - - internalServiceToken: [] + - internalServiceToken: [] requestBody: required: true content: application/json: schema: - $ref: '#/components/schemas/PairedOrder' + $ref: '#/components/schemas/Arm' responses: '200': - description: Diagnostic receipt in the standard ApiResponse wrapper; code is string 200. + description: ApiResponse code 200. status data is null if no grant exists. content: application/json: schema: type: object + additionalProperties: false properties: - code: { type: string, example: '200' } - data: { $ref: '#/components/schemas/Receipt' } - '400': { description: 'Invalid JSON, unknown fields, invalid scope/evidence, duplicate SKU or request limit.' } - '401': { description: Invalid internal service token. } - '409': { description: Session configuration changed or order identity reused with different evidence. } - '503': { description: Internal service token unconfigured or diagnostic storage unavailable. } + code: + type: string + message: + type: string + nullable: true + data: + $ref: '#/components/schemas/View' + nullable: true + required: + - code + - data + '400': + description: Invalid scope, changed frozen evidence, invalid observation or disabled feature. + '401': + description: Missing or invalid internal service token. + '503': + description: Diagnostic admission, dedicated connection or storage unavailable; collection is unaffected. + /internal/lastleg/self-operated-analytics/quantity-diagnostics/status: + post: + operationId: quantityDiagnosticStatus + security: + - internalServiceToken: [] + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/Status' + responses: + '200': + description: ApiResponse code 200. status data is null if no grant exists. + content: + application/json: + schema: + type: object + additionalProperties: false + properties: + code: + type: string + message: + type: string + nullable: true + data: + $ref: '#/components/schemas/View' + nullable: true + required: + - code + - data + '400': + description: Invalid scope, changed frozen evidence, invalid observation or disabled feature. + '401': + description: Missing or invalid internal service token. + '503': + description: Diagnostic admission, dedicated connection or storage unavailable; collection is unaffected. + /internal/lastleg/self-operated-analytics/quantity-diagnostics/observe: + post: + operationId: quantityDiagnosticObserve + security: + - internalServiceToken: [] + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/Event' + responses: + '200': + description: ApiResponse code 200. status data is null if no grant exists. + content: + application/json: + schema: + type: object + additionalProperties: false + properties: + code: + type: string + message: + type: string + nullable: true + data: + $ref: '#/components/schemas/Receipt' + nullable: true + required: + - code + - data + '400': + description: Invalid scope, changed frozen evidence, invalid observation or disabled feature. + '401': + description: Missing or invalid internal service token. + '503': + description: Diagnostic admission, dedicated connection or storage unavailable; collection is unaffected. + /internal/lastleg/self-operated-analytics/quantity-diagnostics/stop: + post: + operationId: quantityDiagnosticStop + security: + - internalServiceToken: [] + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/Stop' + responses: + '200': + description: ApiResponse code 200. status data is null if no grant exists. + content: + application/json: + schema: + type: object + additionalProperties: false + properties: + code: + type: string + message: + type: string + nullable: true + data: + $ref: '#/components/schemas/View' + nullable: true + required: + - code + - data + '400': + description: Invalid scope, changed frozen evidence, invalid observation or disabled feature. + '401': + description: Missing or invalid internal service token. + '503': + description: Diagnostic admission, dedicated connection or storage unavailable; collection is unaffected. components: securitySchemes: internalServiceToken: @@ -42,64 +170,365 @@ components: in: header name: X-Internal-Service-Token schemas: - PairedOrder: + Arm: type: object additionalProperties: false - required: [diagnosticSessionId, collectionSessionId, rootTaskId, sourceAccount, warehouseCode, ownerCode, orderNo, listObservation, detailObservation, skus] properties: - diagnosticSessionId: { type: string, maxLength: 128 } - collectionSessionId: { type: string, maxLength: 128 } - rootTaskId: { type: string, maxLength: 128 } - sourceAccount: { type: string, maxLength: 64 } - warehouseCode: { type: string, maxLength: 64 } - ownerCode: { type: string, maxLength: 128 } - orderNo: { type: string, maxLength: 128 } - listObservation: { $ref: '#/components/schemas/Observation' } - detailObservation: { $ref: '#/components/schemas/Observation' } - skus: + diagnosticSessionId: + type: string + maxLength: 128 + minLength: 1 + sourceAccount: + type: string + maxLength: 64 + minLength: 1 + warehouseCodes: type: array minItems: 1 - maxItems: 200 - description: Union of list/detail SKUs; duplicate source lines aggregated, unknown never zero-filled. - items: { $ref: '#/components/schemas/Sku' } + maxItems: 8 + uniqueItems: true + items: + type: string + maxLength: 64 + minLength: 1 + ownerCode: + type: string + maxLength: 128 + minLength: 1 + description: Explicit owner code, or * for all allowed owners. + excludedRootTaskId: + type: string + maxLength: 128 + minLength: 1 + nullable: true + description: Must match the actual started, unfinished automatic root at arm; null only when none exists. + required: + - diagnosticSessionId + - sourceAccount + - warehouseCodes + - ownerCode + Status: + type: object + additionalProperties: false + properties: + sourceAccount: + type: string + maxLength: 64 + minLength: 1 + producerId: + type: string + maxLength: 128 + minLength: 1 + attach: + type: boolean + description: True only on Integration process attachment; replacement marks delivery uncertain and fences + prior producer. + drained: + type: boolean + description: Only after observing DRAINING, closing intake and draining previously queued callbacks. + required: + - sourceAccount + - producerId + - attach + - drained + Stop: + type: object + additionalProperties: false + properties: + diagnosticSessionId: + type: string + maxLength: 128 + minLength: 1 + sourceAccount: + type: string + maxLength: 64 + minLength: 1 + producerId: + type: string + maxLength: 128 + minLength: 1 + reason: + type: string + enum: + - INCOMPLETE + - QUEUE_FULL + - OBSERVER_FAILED + - SHUTDOWN + required: + - diagnosticSessionId + - sourceAccount + - producerId + - reason Observation: type: object additionalProperties: false - required: [observedAt, attemptId, skuSetComplete] properties: - status: { type: string, nullable: true, maxLength: 32 } + status: + type: string + maxLength: 32 + minLength: 1 + nullable: true outboundTime: type: string + maxLength: 64 + minLength: 1 nullable: true - maxLength: 32 - description: Original Asia/Shanghai source text, null if absent; not reinterpreted as another timezone. - observedAt: { type: string, format: date-time } - attemptId: { type: string, maxLength: 128 } - skuSetComplete: { type: boolean } + observedAt: + type: string + format: date-time + attemptId: + type: string + maxLength: 128 + minLength: 1 + skuSetComplete: + type: boolean + required: + - status + - outboundTime + - observedAt + - attemptId + - skuSetComplete Sku: type: object additionalProperties: false - required: [sku, listPresent, detailPresent] properties: - sku: { type: string, maxLength: 128 } - listQty: { type: number, nullable: true, minimum: 0, multipleOf: 0.000001 } - forecastQty: { type: number, nullable: true, minimum: 0, multipleOf: 0.000001 } - actualQty: { type: number, nullable: true, minimum: 0, multipleOf: 0.000001 } - listPresent: { type: boolean } - detailPresent: { type: boolean } - Receipt: - type: object - required: [outcome, acceptedOrders, acceptedSkuRows, acceptedBytes] - properties: - outcome: { type: string, enum: [ACCEPTED, DUPLICATE, STOPPED] } - comparison: + sku: type: string + maxLength: 128 + minLength: 1 + listQty: + type: number nullable: true - enum: [SKU_SET_MISMATCH, QUANTITY_UNKNOWN, FORECAST_ACTUAL_DIFFERENCE, ALL_EQUAL, LIST_QUANTITY_DIFFERENCE] + minimum: 0 + description: Up to 32 integer digits, at most 6 decimal places; unknown stays null. + forecastQty: + type: number + nullable: true + minimum: 0 + description: Up to 32 integer digits, at most 6 decimal places; unknown stays null. + actualQty: + type: number + nullable: true + minimum: 0 + description: Up to 32 integer digits, at most 6 decimal places; unknown stays null. + required: + - sku + - listQty + - forecastQty + - actualQty + Key: + type: object + additionalProperties: false + properties: + warehouseCode: + type: string + maxLength: 64 + minLength: 1 + ownerCode: + type: string + maxLength: 128 + minLength: 1 + orderNo: + type: string + maxLength: 128 + minLength: 1 + paired: + type: boolean + required: + - warehouseCode + - ownerCode + - orderNo + - paired + Event: + type: object + additionalProperties: false + properties: + diagnosticSessionId: + type: string + maxLength: 128 + minLength: 1 + sourceAccount: + type: string + maxLength: 64 + minLength: 1 + producerId: + type: string + maxLength: 128 + minLength: 1 + rootTaskId: + type: string + maxLength: 128 + minLength: 1 + collectionSessionId: + type: string + maxLength: 128 + minLength: 1 + operation: + type: string + enum: + - OUTBOUND_QUERY + - OUTBOUND_DETAIL + warehouseCode: + type: string + maxLength: 64 + minLength: 1 + ownerCode: + type: string + maxLength: 128 + minLength: 1 + orderNo: + type: string + maxLength: 128 + minLength: 1 + side: + type: string + enum: + - LIST + - DETAIL + observation: + $ref: '#/components/schemas/Observation' + skus: + type: array + maxItems: 200 + items: + $ref: '#/components/schemas/Sku' + description: Empty allowed only when skuSetComplete=false; persists inconclusive evidence. + required: + - diagnosticSessionId + - sourceAccount + - producerId + - rootTaskId + - collectionSessionId + - operation + - warehouseCode + - ownerCode + - orderNo + - side + - observation + - skus + View: + type: object + additionalProperties: false + properties: + diagnosticSessionId: + type: string + maxLength: 128 + minLength: 1 + sourceAccount: + type: string + maxLength: 64 + minLength: 1 + rootTaskId: + type: string + maxLength: 128 + minLength: 1 + nullable: true + collectionSessionId: + type: string + maxLength: 128 + minLength: 1 + nullable: true + state: + type: string + enum: + - ARMED + - BOUND + - DRAINING + - TERMINAL stopReason: type: string nullable: true - enum: [DISABLED, NOT_STARTED, EXPIRED, LIMIT_REACHED, FORECAST_ACTUAL_DIFFERENCE] - acceptedOrders: { type: integer } - acceptedSkuRows: { type: integer } - acceptedBytes: { type: integer, format: int64, description: Canonical whitelist JSON UTF8 bytes persisted in this session. } + startsAt: + type: string + format: date-time + nullable: true + expiresAt: + type: string + format: date-time + nullable: true + bindBefore: + type: string + format: date-time + nullable: true + warehouseCodes: + type: array + maxItems: 8 + items: + type: string + maxLength: 64 + minLength: 1 + ownerCode: + type: string + maxLength: 128 + minLength: 1 + excludedRootTaskId: + type: string + maxLength: 128 + minLength: 1 + nullable: true + candidates: + type: integer + minimum: 0 + maximum: 100 + pairedOrders: + type: integer + minimum: 0 + maximum: 100 + retainedRows: + type: integer + minimum: 0 + maximum: 2000 + retainedBytes: + type: integer + minimum: 0 + maximum: 2097152 + deliveryUncertain: + type: boolean + selected: + type: array + maxItems: 100 + items: + $ref: '#/components/schemas/Key' + required: + - diagnosticSessionId + - sourceAccount + - rootTaskId + - collectionSessionId + - state + - stopReason + - startsAt + - expiresAt + - bindBefore + - warehouseCodes + - ownerCode + - excludedRootTaskId + - candidates + - pairedOrders + - retainedRows + - retainedBytes + - deliveryUncertain + - selected + Receipt: + type: object + additionalProperties: false + properties: + outcome: + type: string + enum: + - ACCEPTED + - IGNORED + - DUPLICATE + - STOPPED + comparison: + type: string + nullable: true + enum: + - ALL_EQUAL + - QUANTITY_DIFFERENCE + - INCONCLUSIVE + session: + $ref: '#/components/schemas/View' + required: + - outcome + - comparison + - session diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticControllerTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticControllerTest.java index 3e36b1723..eb51575c1 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticControllerTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticControllerTest.java @@ -1,47 +1,27 @@ package com.baiyue.lastleg.selfoperated.analytics.diagnostic; - import com.baiyue.lastleg.security.InternalServiceTokenGuard; import com.fasterxml.jackson.databind.ObjectMapper; -import java.nio.charset.StandardCharsets; import org.junit.jupiter.api.Test; import static org.assertj.core.api.Assertions.*; import static org.mockito.Mockito.*; - +import org.springframework.mock.web.MockHttpServletRequest; class QuantityDiagnosticControllerTest { - @Test void httpStatusContractSurvivesSharedBusinessErrorAdvice() throws Exception { - var service = mock(QuantityDiagnosticService.class); - var controller = new QuantityDiagnosticController(new InternalServiceTokenGuard("test-token"), service, new ObjectMapper()); - var mvc = org.springframework.test.web.servlet.setup.MockMvcBuilders.standaloneSetup(controller) - .setControllerAdvice(new com.baiyue.common.web.GlobalExceptionHandler()).build(); - String path = "/internal/lastleg/self-operated-analytics/quantity-diagnostics/orders"; - mvc.perform(org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post(path) - .contentType("application/json").content("{}")) - .andExpect(org.springframework.test.web.servlet.result.MockMvcResultMatchers.status().isUnauthorized()); - mvc.perform(org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post(path) - .header("X-Internal-Service-Token", "test-token").contentType("application/json").content("{\"address\":1}")) - .andExpect(org.springframework.test.web.servlet.result.MockMvcResultMatchers.status().isBadRequest()); - } - @Test void rejectsInvalidTokenBeforeParsingOrStorage() { - var service = mock(QuantityDiagnosticService.class); - var controller = new QuantityDiagnosticController(new InternalServiceTokenGuard("test-token"), service, new ObjectMapper()); - assertThatThrownBy(() -> controller.accept(null, new byte[0])).hasMessageContaining("401"); - assertThatThrownBy(() -> controller.accept("wrong", new byte[0])).hasMessageContaining("401"); - verifyNoInteractions(service); - } - @Test void missingTokenConfigurationFailsClosed() { - var service = mock(QuantityDiagnosticService.class); - var controller = new QuantityDiagnosticController(new InternalServiceTokenGuard(""), service, new ObjectMapper()); - assertThatThrownBy(() -> controller.accept("any", new byte[0])).hasMessageContaining("503"); - verifyNoInteractions(service); - } - @Test void rejectsNonWhitelistedFieldsTrailingDocumentsAndOversizedPayload() { - var service = mock(QuantityDiagnosticService.class); - var controller = new QuantityDiagnosticController(new InternalServiceTokenGuard("test-token"), service, new ObjectMapper()); - assertThatThrownBy(() -> controller.accept("test-token", "{\"address\":\"not allowed\"}".getBytes(StandardCharsets.UTF_8))) - .hasMessageContaining("DIAGNOSTIC_JSON_INVALID"); - assertThatThrownBy(() -> controller.accept("test-token", "{} {}".getBytes(StandardCharsets.UTF_8))) - .hasMessageContaining("DIAGNOSTIC_JSON_INVALID"); - assertThatThrownBy(() -> controller.accept("test-token", new byte[262145])).hasMessageContaining("DIAGNOSTIC_REQUEST_TOO_LARGE"); - verifyNoInteractions(service); - } + @Test void everyRouteRequiresInternalTokenBeforeParsing()throws Exception{ + var service=mock(QuantityDiagnosticService.class);var controller=new QuantityDiagnosticController(new InternalServiceTokenGuard("test-token"),service,new ObjectMapper()); + var mvc=org.springframework.test.web.servlet.setup.MockMvcBuilders.standaloneSetup(controller).setControllerAdvice(new com.baiyue.common.web.GlobalExceptionHandler()).build(); + for(String action:new String[]{"arm","status","observe","stop"})mvc.perform(org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post("/internal/lastleg/self-operated-analytics/quantity-diagnostics/"+action).contentType("application/json").content("{}")) + .andExpect(org.springframework.test.web.servlet.result.MockMvcResultMatchers.status().isUnauthorized()); + verifyNoInteractions(service); + } + @Test void strictWhitelistAndBoundedBody(){ + var service=mock(QuantityDiagnosticService.class);var c=new QuantityDiagnosticController(new InternalServiceTokenGuard("test-token"),service,new ObjectMapper()); + for(String body:new String[]{"{\"address\":1}","{} {}"}){var request=new MockHttpServletRequest();request.setContent(body.getBytes(java.nio.charset.StandardCharsets.UTF_8));assertThatThrownBy(()->c.observe("test-token",request)).hasMessageContaining("JSON_INVALID");} + var request=new MockHttpServletRequest();request.setContent(new byte[262145]);assertThatThrownBy(()->c.observe("test-token",request)).hasMessageContaining("REQUEST_TOO_LARGE");verifyNoInteractions(service); + } + @Test void streamIsLimitedEvenWithoutContentLength()throws Exception{ + var service=mock(QuantityDiagnosticService.class);var c=new QuantityDiagnosticController(new InternalServiceTokenGuard("test-token"),service,new ObjectMapper()); + var request=mock(jakarta.servlet.http.HttpServletRequest.class);when(request.getContentLengthLong()).thenReturn(-1L); + var count=new java.util.concurrent.atomic.AtomicInteger();var stream=new jakarta.servlet.ServletInputStream(){public int read(){count.incrementAndGet();return 32;}public boolean isFinished(){return false;}public boolean isReady(){return true;}public void setReadListener(jakarta.servlet.ReadListener listener){}}; + when(request.getInputStream()).thenReturn(stream);assertThatThrownBy(()->c.observe("test-token",request)).hasMessageContaining("REQUEST_TOO_LARGE");assertThat(count.get()).isEqualTo(262145);verifyNoInteractions(service); + } } diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticServiceTest.java index d0ab0ab8b..debf7a3f9 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticServiceTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticServiceTest.java @@ -1,5 +1,4 @@ package com.baiyue.lastleg.selfoperated.analytics.diagnostic; - import static com.baiyue.lastleg.selfoperated.analytics.diagnostic.QuantityDiagnosticContract.*; import static org.assertj.core.api.Assertions.*; import com.fasterxml.jackson.databind.ObjectMapper; @@ -12,158 +11,133 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.springframework.core.io.ClassPathResource; import org.springframework.jdbc.core.JdbcTemplate; -import org.springframework.jdbc.datasource.DataSourceTransactionManager; -import org.springframework.jdbc.datasource.DriverManagerDataSource; -import org.springframework.web.server.ResponseStatusException; - +import org.springframework.jdbc.datasource.*; class QuantityDiagnosticServiceTest { - private final Instant start = Instant.parse("2026-10-13T15:00:00Z"); - private final ObjectMapper json = new ObjectMapper().findAndRegisterModules(); - private JdbcTemplate jdbc; - private DataSourceTransactionManager manager; - private QuantityDiagnosticService service; - private QuantityDiagnosticProperties properties; - - @BeforeEach void prepare() throws Exception { - var ds = new DriverManagerDataSource("jdbc:h2:mem:" + UUID.randomUUID() + ";MODE=MySQL;DB_CLOSE_DELAY=-1;LOCK_TIMEOUT=2000", "sa", ""); - jdbc = new JdbcTemplate(ds); manager = new DataSourceTransactionManager(ds); - // Exercise the dedicated production DDL, omitting only MySQL table options/comments. - String ddl; - try (var in = new ClassPathResource("db/selfop-quantity-diagnostics.sql").getInputStream()) { - ddl = new String(in.readAllBytes(), StandardCharsets.UTF_8); - } - ddl = ddl.replaceAll("(?m)^--.*$", "").replaceAll("COMMENT\\s*=?\\s*'[^']*'", "") - .replaceAll("ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin", ""); - for (String statement : ddl.split(";")) if (!statement.isBlank()) jdbc.execute(statement); - properties = config(start.plusSeconds(1800), true); - service = instance(properties, start.plusSeconds(10)); - } - private QuantityDiagnosticProperties config(Instant expires, boolean enabled) { - return new QuantityDiagnosticProperties(enabled, "diag-1", "collection-1", "root-1", "account-1", - List.of("warehouse-1"), start, expires); - } - private QuantityDiagnosticService instance(QuantityDiagnosticProperties p, Instant now) { - return new QuantityDiagnosticService(jdbc, manager, json, p, Clock.fixed(now, ZoneOffset.UTC)); - } - private Sku sku(String code, String list, String forecast, String actual) { - return new Sku(code, list == null ? null : new BigDecimal(list), forecast == null ? null : new BigDecimal(forecast), - actual == null ? null : new BigDecimal(actual), true, true); - } - private Order order(String no, Sku... skus) { - return new Order("diag-1", "collection-1", "root-1", "account-1", "warehouse-1", "owner-1", no, - new Observation("100", "2026-10-06 12:00:00", start.plusSeconds(1), "list-1", true), - new Observation("100", "2026-10-06 12:00:00", start.plusSeconds(2), "detail-1", true), List.of(skus)); - } - @Test void firstDifferencePersistsEvidenceAndStopsFurtherOrdersIncludingAfterRestart() { - var request = order("order-1", sku("sku-1", "10", "10", "8")); - var result = service.accept(request); - assertThat(result.outcome()).isEqualTo("ACCEPTED"); - assertThat(result.comparison()).isEqualTo(Comparison.FORECAST_ACTUAL_DIFFERENCE); - assertThat(result.stopReason()).isEqualTo("FORECAST_ACTUAL_DIFFERENCE"); - service = instance(properties, start.plusSeconds(20)); - assertThat(service.accept(request).outcome()).isEqualTo("DUPLICATE"); - assertThat(service.accept(order("order-2", sku("sku-1", "1", "1", "1"))).outcome()).isEqualTo("STOPPED"); - assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ll_selfop_quantity_diagnostic_order", Integer.class)).isEqualTo(1); - } - @Test void listMatchingActualStillRecordsForecastDifference() { - assertThat(service.accept(order("order-1", sku("sku", "8", "10", "8"))).comparison()) - .isEqualTo(Comparison.FORECAST_ACTUAL_DIFFERENCE); - } - @Test void idempotencyCanonicalizesDecimalsAndRowOrderButRejectsChangedEvidence() { - service.accept(order("order-1", sku("b", "1.0", "1", "1"), sku("a", "2", "2", "2"))); - var result = service.accept(order("order-1", sku("a", "2.00", "2", "2"), sku("b", "1", "1", "1"))); - assertThat(result.outcome()).isEqualTo("DUPLICATE"); - assertThat(result.acceptedOrders()).isEqualTo(1); - assertThatThrownBy(() -> service.accept(order("order-1", sku("a", "3", "3", "3")))) - .isInstanceOf(ResponseStatusException.class).hasMessageContaining("409"); - } - @Test void skuReplacementCannotBeHiddenByEqualTotalsAndDifferenceStillStops() { - var result = service.accept(order("order-1", - new Sku("old", BigDecimal.TEN, null, null, true, false), - new Sku("new", null, BigDecimal.TEN, BigDecimal.valueOf(8), false, true))); - assertThat(result.comparison()).isEqualTo(Comparison.SKU_SET_MISMATCH); - assertThat(result.stopReason()).isEqualTo("FORECAST_ACTUAL_DIFFERENCE"); - } - @Test void missingQuantityRemainsUnknown() { - var result = service.accept(order("order-1", sku("sku", null, "1", "1"))); - assertThat(result.comparison()).isEqualTo(Comparison.QUANTITY_UNKNOWN); - assertThat(result.stopReason()).isNull(); - } - @Test void duplicateSkusAndCrossGenerationAreRejectedWithoutWrites() { - assertThatThrownBy(() -> service.accept(order("order-1", sku("sku", "1", "1", "1"), sku("sku", "1", "1", "1")))) - .hasMessageContaining("DIAGNOSTIC_DUPLICATE_SKU"); - Order r = order("order-1", sku("sku", "1", "1", "1")); - assertThatThrownBy(() -> service.accept(new Order(r.diagnosticSessionId(), "other-generation", r.rootTaskId(), r.sourceAccount(), - r.warehouseCode(), r.ownerCode(), r.orderNo(), r.listObservation(), r.detailObservation(), r.skus()))) - .hasMessageContaining("DIAGNOSTIC_SCOPE_MISMATCH"); - assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ll_selfop_quantity_diagnostic_session", Integer.class)).isZero(); - } - @Test void expiredAndReconfiguredSessionsCannotResetCounters() { - var r = order("order-1", sku("sku", "1", "1", "1")); service.accept(r); - var expired = instance(properties, start.plusSeconds(1800)); - assertThat(expired.accept(order("order-2", sku("sku", "1", "1", "1"))).stopReason()).isEqualTo("EXPIRED"); - assertThat(expired.accept(r).outcome()).isEqualTo("DUPLICATE"); - assertThatThrownBy(() -> instance(config(start.plusSeconds(1799), true), start.plusSeconds(20)).accept(r)) - .hasMessageContaining("DIAGNOSTIC_SESSION_CONFIGURATION_CHANGED"); - } - @Test void disabledRequiresNoTablesOrValidSession() { - jdbc.execute("DROP TABLE ll_selfop_quantity_diagnostic_order"); - jdbc.execute("DROP TABLE ll_selfop_quantity_diagnostic_session"); - assertThat(instance(config(start.plusSeconds(1800), false), start).accept(null).stopReason()).isEqualTo("DISABLED"); - } - @Test void concurrentRequestsCannotExceedHundredOrders() throws Exception { - var pool = Executors.newFixedThreadPool(4); - try { - List> jobs = new ArrayList<>(); - for (int i = 0; i < 104; i++) { final int n = i; jobs.add(() -> service.accept(order("order-" + n, sku("sku", "1", "1", "1")))); } - var results = pool.invokeAll(jobs); - int accepted = 0; - for (var result : results) if (result.get().outcome().equals("ACCEPTED")) accepted++; - assertThat(accepted).isEqualTo(100); - assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ll_selfop_quantity_diagnostic_order", Integer.class)).isEqualTo(100); - assertThat(instance(properties, start.plusSeconds(20)).accept(order("after-restart", sku("sku", "1", "1", "1"))).stopReason()) - .isEqualTo("LIMIT_REACHED"); - } finally { pool.shutdownNow(); } - } - @Test void rowLimitAndFailedOrderInsertDoNotPartiallyAdvanceCounters() { - service.accept(order("order-1", sku("sku", "1", "1", "1"))); - jdbc.update("UPDATE ll_selfop_quantity_diagnostic_session SET accepted_sku_rows=2000"); - assertThat(service.accept(order("order-2", sku("sku", "1", "1", "1"))).stopReason()).isEqualTo("LIMIT_REACHED"); - assertThat(jdbc.queryForObject("SELECT accepted_orders FROM ll_selfop_quantity_diagnostic_session", Integer.class)).isEqualTo(1); - } - @Test void byteLimitCannotBeResetByRestartOrPartiallySaveOrder() { - service.accept(order("order-1", sku("sku", "1", "1", "1"))); - jdbc.update("UPDATE ll_selfop_quantity_diagnostic_session SET accepted_bytes=2097150"); - var receipt = instance(properties, start.plusSeconds(20)).accept(order("order-2", sku("sku", "1", "1", "1"))); - assertThat(receipt.stopReason()).isEqualTo("LIMIT_REACHED"); - assertThat(receipt.acceptedBytes()).isEqualTo(2097150); - assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ll_selfop_quantity_diagnostic_order", Integer.class)).isEqualTo(1); - } - @Test void dedicatedDdlHasChineseCommentsOnEveryTableAndOrdinaryColumn() throws Exception { - String ddl; - try (var in = new ClassPathResource("db/selfop-quantity-diagnostics.sql").getInputStream()) { - ddl = new String(in.readAllBytes(), StandardCharsets.UTF_8); - } - var fields = java.util.regex.Pattern.compile("(?m)^ [a-z][a-z_]+ (?:VARCHAR|CHAR|INT|BIGINT|JSON)\\b[^\\n]*").matcher(ddl); - int count = 0; - while (fields.find()) { - assertThat(fields.group()).matches("(?s).*COMMENT '[^']*[\\u4e00-\\u9fff][^']*'.*"); count++; - } - assertThat(count).isEqualTo(23); - assertThat(java.util.regex.Pattern.compile("COMMENT='[^']*[\\u4e00-\\u9fff][^']*';").matcher(ddl).results().count()).isEqualTo(2); - } - @Test void storageFailureRollsBackSessionCounterAndSessionCreation() { - jdbc.execute("ALTER TABLE ll_selfop_quantity_diagnostic_order ADD CONSTRAINT reject_test_order CHECK(order_no <> 'bad-order')"); - assertThatThrownBy(() -> service.accept(order("bad-order", sku("sku", "1", "1", "1")))) - .isInstanceOf(org.springframework.dao.DataIntegrityViolationException.class); - assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ll_selfop_quantity_diagnostic_session", Integer.class)).isZero(); - } - @Test void cancellationAndMissingOutboundTimeRemainEvidenceOnly() { - var r = order("cancelled", sku("sku", "1", "1", "1")); - var cancelled = new Observation("cancelled", null, start.plusSeconds(2), "detail-cancel", true); - var result = service.accept(new Order(r.diagnosticSessionId(), r.collectionSessionId(), r.rootTaskId(), r.sourceAccount(), - r.warehouseCode(), r.ownerCode(), r.orderNo(), r.listObservation(), cancelled, r.skus())); - assertThat(result.comparison()).isEqualTo(Comparison.ALL_EQUAL); - assertThat(result.acceptedOrders()).isEqualTo(1); - } + final Instant start=Instant.parse("2026-10-14T00:00:00Z"); + final ObjectMapper json=new ObjectMapper().findAndRegisterModules(); + JdbcTemplate jdbc;DataSourceTransactionManager manager;QuantityDiagnosticService service;MutableClock clock; + final QuantityDiagnosticProperties config=new QuantityDiagnosticProperties(true,"account",List.of("W1","W2")); + @BeforeEach void setup()throws Exception{ + var ds=new DriverManagerDataSource("jdbc:h2:mem:"+UUID.randomUUID()+";MODE=MySQL;DB_CLOSE_DELAY=-1;LOCK_TIMEOUT=2000","sa","");jdbc=new JdbcTemplate(ds);manager=new DataSourceTransactionManager(ds);clock=new MutableClock(start); + String ddl;try(var in=new ClassPathResource("db/selfop-quantity-diagnostics.sql").getInputStream()){ddl=new String(in.readAllBytes(),StandardCharsets.UTF_8);} + ddl=ddl.replaceAll("(?m)^--.*$","").replaceAll("COMMENT\\s*=?\\s*'[^']*'", "").replace("evidence_json JSON", "evidence_json LONGTEXT").replace("ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin","");for(String sql:ddl.split(";"))if(!sql.isBlank())jdbc.execute(sql); + jdbc.execute("CREATE TABLE ll_selfop_sync_task(id BIGINT PRIMARY KEY,source_account VARCHAR(64),execution_purpose VARCHAR(32),upstream_task_id BIGINT,validation_root_task_id BIGINT,requested_by VARCHAR(64),scope_hash VARCHAR(64),request_key VARCHAR(128),task_role VARCHAR(32),task_type VARCHAR(32),discovery_scope_id BIGINT,query_from_utc TIMESTAMP,query_until_utc TIMESTAMP,query_scope_json VARCHAR(2000),status VARCHAR(32),attempt_count INT DEFAULT 0)"); + for(long id=1;id<=3;id++)jdbc.update("INSERT INTO ll_selfop_sync_task VALUES(?,?,?,NULL,NULL,?,?,?,?,?,?,?, ?,?,?,0)",id,"account","BUSINESS","SELFOP_CONTINUOUS_PLANNER","a".repeat(64),"continuous:outbound:2026-10-10:"+"a".repeat(12),"ACCOUNT_DISCOVERY","BACKFILL",1L,LocalDateTime.parse("2026-10-09T16:00:00"),LocalDateTime.parse("2026-10-10T16:00:00"),"{\"startTime\":\"2026-10-10T00:00\",\"endTime\":\"2026-10-11T00:00\"}",id==1?"RUNNING":"READY"); + service=instance();service.arm(arm("diag"));service.status(new Status("account","producer",true,false));clock.now=start.plusSeconds(10); + } + QuantityDiagnosticService instance(){return new QuantityDiagnosticService(jdbc,manager,json,config,clock);} + Arm arm(String id){return new Arm(id,"account",List.of("W1","W2"),"*","1");} + Event event(String root,String order,String side,BigDecimal list,BigDecimal forecast,BigDecimal actual){return new Event("diag","account","producer",root,"selfop-"+root,"LIST".equals(side)?"OUTBOUND_QUERY":"OUTBOUND_DETAIL","W1","owner",order,side,new Observation("100",null,clock.instant(),"attempt",true),List.of(new Sku("SKU",list,forecast,actual)));} + Event list(String order){return event("2",order,"LIST",BigDecimal.TEN,null,null);} + Event detail(String order){return event("2",order,"DETAIL",null,BigDecimal.TEN,BigDecimal.TEN);} + View status(){return service.status(new Status("account","producer",false,false));} + @Test void explicitArmIsIdempotentAndBindsOnlyNextEligibleList(){ + assertThat(service.arm(arm("diag")).startsAt()).isEqualTo(start); + assertThat(service.observe(detail("early")).outcome()).isEqualTo("IGNORED"); + assertThat(service.observe(event("1","old","LIST",BigDecimal.ONE,null,null)).outcome()).isEqualTo("IGNORED"); + Event locate=list("locate");assertThatThrownBy(()->service.observe(new Event(locate.diagnosticSessionId(),locate.sourceAccount(),locate.producerId(),locate.rootTaskId(),locate.collectionSessionId(),"OUTBOUND_LOCATE",locate.warehouseCode(),locate.ownerCode(),locate.orderNo(),locate.side(),locate.observation(),locate.skus()))).hasMessageContaining("OPERATION"); + View bound=service.observe(list("a")).session();assertThat(bound.rootTaskId()).isEqualTo("2");assertThat(bound.expiresAt()).isEqualTo(clock.instant().plusSeconds(14400)); + assertThatThrownBy(()->service.observe(event("3","other","LIST",BigDecimal.ONE,null,null))).hasMessageContaining("ROOT_FROZEN"); + assertThatThrownBy(()->service.arm(arm("second"))).hasMessageContaining("ALREADY_ARMED"); + } + @Test void durablePendingSurvivesReceiverRestartAndPreservesOriginalAttempt(){ + Event original=list("a");service.observe(original);service=instance();clock.now=start.plusSeconds(5000); + assertThat(service.observe(original).outcome()).isEqualTo("DUPLICATE"); + var result=service.observe(detail("a"));assertThat(result.comparison()).isEqualTo("ALL_EQUAL");assertThat(result.session().candidates()).isEqualTo(1); + String payload=jdbc.queryForObject("SELECT evidence_json FROM ll_selfop_quantity_diagnostic_order",String.class);assertThat(payload).contains("attempt").contains("00:00:10Z"); + } + @Test void anyKnownInequalityStopsIncludingListDifferenceAndTerminalReplay(){ + Event l=list("a");service.observe(l);Event d=event("2","a","DETAIL",null,BigDecimal.ONE,BigDecimal.ONE); + var result=service.observe(d);assertThat(result.comparison()).isEqualTo("QUANTITY_DIFFERENCE");assertThat(result.session().state()).isEqualTo("TERMINAL"); + assertThat(instance().observe(d).outcome()).isEqualTo("DUPLICATE");assertThat(service.observe(list("b")).outcome()).isEqualTo("STOPPED"); + } + @Test void unknownListStopsAndRetainsEvidence(){ + var receipt=service.observe(event("2","a","LIST",null,null,null));assertThat(receipt.comparison()).isEqualTo("INCONCLUSIVE");assertThat(receipt.session().stopReason()).isEqualTo("INCONCLUSIVE"); + assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ll_selfop_quantity_diagnostic_pending",Integer.class)).isEqualTo(1); + } + @Test void missingSkuAndUnknownDetailStopAtomically(){ + service.observe(list("a"));Event d=detail("a");var mismatch=new Event(d.diagnosticSessionId(),d.sourceAccount(),d.producerId(),d.rootTaskId(),d.collectionSessionId(),d.operation(),d.warehouseCode(),d.ownerCode(),d.orderNo(),d.side(),d.observation(),List.of(new Sku("OTHER",null,null,BigDecimal.TEN))); + assertThat(service.observe(mismatch).comparison()).isEqualTo("INCONCLUSIVE");assertThat(status().state()).isEqualTo("TERMINAL");assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ll_selfop_quantity_diagnostic_order",Integer.class)).isEqualTo(1); + } + @Test void lateCallbackCannotBindEvenWhenObservationWasBeforeDeadline(){ + Event e=list("a");clock.now=start.plusSeconds(21600);assertThat(service.observe(e).session().stopReason()).isEqualTo("BIND_TIMEOUT");assertThat(status().rootTaskId()).isNull(); + } + @Test void latestPossibleBindStillCannotExtendAbsoluteTenHours(){ + clock.now=start.plusSeconds(21599);assertThat(service.observe(list("a")).session().expiresAt()).isEqualTo(start.plusSeconds(35999)); + clock.now=start.plusSeconds(36000);assertThat(service.observe(detail("a")).session().stopReason()).isEqualTo("INCOMPLETE"); + } + @Test void rootEndDrainsOnlyAcceptedCandidatesAndNeverExtendsDeadline(){ + service.observe(list("a"));Instant expires=status().expiresAt();jdbc.update("UPDATE ll_selfop_sync_task SET status='SUCCEEDED' WHERE id=2"); + assertThat(status().state()).isEqualTo("DRAINING");assertThat(service.observe(list("new")).outcome()).isEqualTo("IGNORED"); + assertThat(service.observe(detail("a")).comparison()).isEqualTo("ALL_EQUAL"); + View done=service.status(new Status("account","producer",false,true));assertThat(done.stopReason()).isEqualTo("ROOT_ENDED_INSUFFICIENT");assertThat(done.expiresAt()).isEqualTo(expires); + } + @Test void producerRestartMarksLostTailUncertainButRestoresPending(){ + service.observe(list("a"));View recovered=service.status(new Status("account","new-producer",true,false));assertThat(recovered.deliveryUncertain()).isTrue();assertThat(recovered.selected()).hasSize(1); + Event d=detail("a");var resumed=new Event(d.diagnosticSessionId(),d.sourceAccount(),"new-producer",d.rootTaskId(),d.collectionSessionId(),d.operation(),d.warehouseCode(),d.ownerCode(),d.orderNo(),d.side(),d.observation(),d.skus());assertThat(service.observe(resumed).comparison()).isEqualTo("ALL_EQUAL"); + jdbc.update("UPDATE ll_selfop_sync_task SET status='SUCCEEDED' WHERE id=2");assertThat(service.status(new Status("account","new-producer",false,true)).stopReason()).isEqualTo("INCOMPLETE"); + } + @Test void candidatesAreCumulativeAndNeverReplaced(){ + for(int i=0;i<100;i++)service.observe(list("o"+i));assertThat(service.observe(list("extra")).outcome()).isEqualTo("IGNORED"); + for(int i=0;i<100;i++)service.observe(detail("o"+i));View done=status();assertThat(done.candidates()).isEqualTo(100);assertThat(done.pairedOrders()).isEqualTo(100);assertThat(done.retainedRows()).isEqualTo(200);assertThat(done.stopReason()).isEqualTo("COMPLETED"); + } + @Test void changedAttemptCannotOverwriteFirstSide(){ + Event a=list("a");service.observe(a);clock.now=start.plusSeconds(20);assertThatThrownBy(()->service.observe(list("a"))).hasMessageContaining("OBSERVATION_CONFLICT");assertThat(status().candidates()).isEqualTo(1); + } + @Test void concurrentReceiversSerializeSameCandidate()throws Exception{ + var other=instance();Event e=list("a");var executor=Executors.newFixedThreadPool(2);try{var one=executor.submit(()->service.observe(e));var two=executor.submit(()->other.observe(e));assertThat(List.of(one.get().outcome(),two.get().outcome())).containsExactlyInAnyOrder("ACCEPTED","DUPLICATE");}finally{executor.shutdownNow();}assertThat(status().candidates()).isEqualTo(1); + } + @Test void businessRootPolicyIsAuthoritative(){ + jdbc.update("UPDATE ll_selfop_sync_task SET requested_by='manual' WHERE id=2");assertThatThrownBy(()->service.observe(list("a"))).hasMessageContaining("ROOT_REJECTED");assertThat(status().rootTaskId()).isNull(); + } + @Test void delayedFirstObservationDoesNotExtendSamplingWindow(){ + Event first=list("a");clock.now=start.plusSeconds(5000);View bound=service.observe(first).session();assertThat(bound.startsAt()).isEqualTo(first.observation().observedAt());assertThat(bound.expiresAt()).isEqualTo(start.plusSeconds(14410)); + } + @Test void endedOrChangedRootCannotBePresentedAsCurrent(){ + service.stop(new Stop("diag","account","producer","INCOMPLETE"));jdbc.update("UPDATE ll_selfop_sync_task SET status='SUCCEEDED' WHERE id=1"); + assertThatThrownBy(()->service.arm(arm("next"))).hasMessageContaining("CURRENT_ROOT_CHANGED"); + assertThat(service.arm(new Arm("next","account",List.of("W1"),"*",null)).excludedRootTaskId()).isNull(); + } + @Test void ambiguousCurrentRootsRefuseArm(){ + service.stop(new Stop("diag","account","producer","INCOMPLETE"));jdbc.update("UPDATE ll_selfop_sync_task SET status='RUNNING' WHERE id=2"); + assertThatThrownBy(()->service.arm(arm("next"))).hasMessageContaining("CURRENT_ROOT_AMBIGUOUS"); + } + @Test void diagnosticDatasourceInitializationFailureDoesNotBreakServiceConstruction(){ + var unavailable=new QuantityDiagnosticService(new DriverManagerDataSource(),json,config); + assertThatThrownBy(()->unavailable.status(new Status("account","p",true,false))).hasMessageContaining("DIAGNOSTIC_DATASOURCE_UNAVAILABLE");unavailable.close(); + } + @Test void fourHourExpiredDelayedFirstListIsNotAccepted(){ + Event first=list("a");clock.now=start.plusSeconds(15000);var r=service.observe(first);assertThat(r.outcome()).isEqualTo("STOPPED");assertThat(r.session().candidates()).isZero(); + } + @Test void rowAndByteCapsNeverRefundOrAcceptOverLimit()throws Exception{ + service.observe(list("a"));jdbc.update("UPDATE ll_selfop_quantity_diagnostic_session SET accepted_sku_rows=2000"); + assertThat(service.observe(detail("a")).session().stopReason()).isEqualTo("INCOMPLETE");assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ll_selfop_quantity_diagnostic_pending",Integer.class)).isEqualTo(1); + jdbc.update("UPDATE ll_selfop_quantity_diagnostic_session SET accepted_sku_rows=1,accepted_bytes=2097152,stop_reason=NULL,active_slot=1"); + assertThat(service.observe(detail("a")).outcome()).isEqualTo("STOPPED");assertThat(status().retainedBytes()).isEqualTo(2097152); + } + @Test void concurrentLastCandidateCannotExceedOneHundred()throws Exception{ + for(int i=0;i<99;i++)service.observe(list("o"+i));var other=instance();var pool=Executors.newFixedThreadPool(2); + try{var a=pool.submit(()->service.observe(list("last-a")));var b=pool.submit(()->other.observe(list("last-b")));assertThat(List.of(a.get().outcome(),b.get().outcome())).containsExactlyInAnyOrder("ACCEPTED","IGNORED");}finally{pool.shutdownNow();}assertThat(status().candidates()).isEqualTo(100); + } + @Test void admissionRejectsBeforeSecondConnectionAcquisition()throws Exception{ + var tm=org.mockito.Mockito.mock(org.springframework.transaction.PlatformTransactionManager.class);var entered=new CountDownLatch(1);var release=new CountDownLatch(1); + org.mockito.Mockito.when(tm.getTransaction(org.mockito.ArgumentMatchers.any())).thenAnswer(call->{entered.countDown();release.await(2,TimeUnit.SECONDS);return new org.springframework.transaction.support.SimpleTransactionStatus();}); + var bounded=new QuantityDiagnosticService(jdbc,tm,json,config,clock);var pool=Executors.newSingleThreadExecutor(); + try{var first=pool.submit(()->bounded.status(new Status("account","producer",false,false)));assertThat(entered.await(1,TimeUnit.SECONDS)).isTrue();assertThatThrownBy(()->bounded.status(new Status("account","producer",false,false))).hasMessageContaining("DIAGNOSTIC_BUSY");release.countDown();first.get();org.mockito.Mockito.verify(tm,org.mockito.Mockito.times(1)).getTransaction(org.mockito.ArgumentMatchers.any());}finally{release.countDown();pool.shutdownNow();} + } + @org.junit.jupiter.params.ParameterizedTest + @org.junit.jupiter.params.provider.ValueSource(strings={"10","8"}) + void forecastTenActualEightStopsRegardlessOfListMeaning(String listQuantity){ + service.observe(event("2","a","LIST",new BigDecimal(listQuantity),null,null));assertThat(service.observe(event("2","a","DETAIL",null,BigDecimal.TEN,new BigDecimal("8"))).comparison()).isEqualTo("QUANTITY_DIFFERENCE"); + } + @Test void historicalEvidenceRemainsUntouchedAndCannotRearmItsIdentifier(){ + jdbc.update("INSERT INTO ll_selfop_quantity_diagnostic_session(diagnostic_session_id,configuration_hash,source_account,collection_session_id,root_task_id,starts_at,expires_at,accepted_orders,accepted_sku_rows,accepted_bytes,stop_reason) VALUES('old',?,'account','old-session','1',?,?,24,25,1234,'EXPIRED')","a".repeat(64),start.toString(),start.toString()); + assertThatThrownBy(()->service.arm(arm("old"))).hasMessageContaining("HISTORICAL_ID_READ_ONLY");assertThat(status().diagnosticSessionId()).isEqualTo("diag");assertThat(jdbc.queryForObject("SELECT accepted_orders FROM ll_selfop_quantity_diagnostic_session WHERE diagnostic_session_id='old'",Integer.class)).isEqualTo(24); + } + @Test void existingStopReasonAndCountersAreTheOnlyPersistedFacts(){ + service.observe(list("a"));service.observe(event("2","a","DETAIL",null,BigDecimal.TEN,BigDecimal.ONE)); + assertThat(jdbc.queryForObject("SELECT stop_reason FROM ll_selfop_quantity_diagnostic_session",String.class)).isEqualTo("QUANTITY_DIFFERENCE");assertThat(jdbc.queryForObject("SELECT accepted_orders FROM ll_selfop_quantity_diagnostic_session",Integer.class)).isEqualTo(1); + String metadata=jdbc.queryForObject("SELECT state_json FROM ll_selfop_quantity_diagnostic_session",String.class);assertThat(metadata).doesNotContain("\"reason\"","\"paired\"","\"bytes\"","\"expiresAt\"");assertThat(instance().status(new Status("account","producer",false,false)).stopReason()).isEqualTo("QUANTITY_DIFFERENCE"); + } + static class MutableClock extends Clock {Instant now;MutableClock(Instant now){this.now=now;}public ZoneId getZone(){return ZoneOffset.UTC;}public Clock withZone(ZoneId zone){return this;}public Instant instant(){return now;}} }