From edb13450a64cd3553885e9f5a2bf46f9441d36bf Mon Sep 17 00:00:00 2001 From: zgw <13623675548@163.com> Date: Wed, 9 Sep 2026 10:00:43 +0800 Subject: [PATCH] fix(shadow): recover completed read timeouts and pair issued observations --- ...shadow-20260909-recovery-pairing-repair.md | 80 +++++ .../evidence/DeferredShadowComparison.java | 205 ++++++++++++ .../shadow/evidence/ShadowCycleEvidence.java | 7 + .../ShadowCycleEvidenceCollector.java | 45 ++- .../evidence/ShadowDifferenceCategory.java | 3 +- .../facts/SqlServerShadowFactReader.java | 4 + .../ruoyi/shadow/kernel/Ft546ShadowCycle.java | 19 +- .../shadow/kernel/Ft546ShadowScheduler.java | 40 ++- .../shadow/startup/ShadowRuntimeIdentity.java | 2 +- .../shadow/Ft546ShadowApplicationTest.java | 2 +- .../shadow/Ft546ShadowFaultIsolationTest.java | 149 ++++++++- .../JdbcShadowEvidenceRepositoryTest.java | 27 ++ .../ShadowDeferredComparisonTest.java | 311 ++++++++++++++++++ .../kernel/ShadowCurrentFieldReplayTest.java | 91 +++++ 14 files changed, 948 insertions(+), 37 deletions(-) create mode 100644 docs/runbooks/ft546-shadow-20260909-recovery-pairing-repair.md create mode 100644 ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/evidence/DeferredShadowComparison.java create mode 100644 ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/evidence/ShadowDeferredComparisonTest.java create mode 100644 ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/kernel/ShadowCurrentFieldReplayTest.java diff --git a/docs/runbooks/ft546-shadow-20260909-recovery-pairing-repair.md b/docs/runbooks/ft546-shadow-20260909-recovery-pairing-repair.md new file mode 100644 index 0000000..3be3026 --- /dev/null +++ b/docs/runbooks/ft546-shadow-20260909-recovery-pairing-repair.md @@ -0,0 +1,80 @@ +# 2026-09-09 影子采集恢复与延迟比较修复 + +适用现场源码基线：`af8b5f1`，分支 `optimize/wcs`，目录 `C:\project\FT546-WCS-SHANDOW`。 + +本次交付供影子程序继续采样验证，不能据此切换正式调度。代码变更范围为 `ruoyi-wcs-shadow`；不改正式调度模块、生产数据、生产只读视图、数据库权限、PLC/WMS 通讯路径。本次无需执行 SQL 迁移或清理数据。 + +新运行标签：`ft546-shadow-20260909-recovery-pairing-v3`。 + +## 修复内容 + +### 完成的读取超时可以恢复 + +旧版本一次 `SOURCE_BATCH_DEADLINE_EXCEEDED` 或 `SOURCE_QUERY_TIMEOUT` 就永久停止采集。新版本确认读取线程及其源库事务已退出后，按 1、2、4、8、30 秒退避；后续成功读取可恢复 `READY`。调度仍遵守原配置的固定轮询间隔，退避期间不查询源库。 + +以下情况仍停止采集：整体执行超时或线程中断而不能确认工作已结束、源库事务提交/回滚失败、检测到重叠读取、影子证据存储或运行状态写入失败、容量保护触发。不会通过启动额外工作线程绕过保护。 + +初始运行状态发布单列为 `RUNTIME_PUBLICATION`。源库读取额度从实际执行读取时开始计算，最多 4 秒，并受整个周期剩余预算约束。默认总预算仍为 10 秒；状态发布已花费 7 秒时最多只剩 3 秒，不再重新给执行线程完整的 10 秒。初始同步状态写入仍由证据库 JDBC 自身的连接/查询超时约束；返回后会检查是否已耗尽周期预算。 + +### 将预测与随后出现的同阶段旧命令比较 + +待下发时的旧 `path_info` 可能属于前一阶段，因此原预测记录及其 `SOURCE_GAP` 必须保留。新程序在后续观测记录上追加 `shadow_difference.category = DEFERRED_COMPARISON`，不修改旧记录和旧哈希。 + +配对要求： + +- 前一成功采样已将该阶段预测提交到影子证据库；后一采样看到普通阶段由 `issued=0 / execution_status=1` 转为 `1 / 3`。 +- GUID、创建时间、任务号、任务类型、起终点、楼层、绑定车辆、货位、优先级及 default02～default08 保持一致。空字符串与 null 的非实质差别统一处理。 +- 两次源库采样时间顺序正确、间隔不超过 30 秒，地图内容不变，活动队列 GUID 没有重复歧义。 +- 旧命令路径可按现有协议投影，并且其起点仍对应预测时的阶段起点；无法证明时记录不可比较。 + +该比较只证明两次队列观测之间的关系，不是 PLC 接收确认，也不是在同一瞬时状态下执行两种算法的严格对照。交通状态可能在采样间变化，差异仍需逐条解释。 + +| reason_code | 含义 | +| --- | --- | +| `DEFERRED_MATCH` | 该次有效配对的车辆、路径、队列字段及虚拟协议写入一致 | +| `DEFERRED_DIFFERENCE` | 有可用配对，但至少存在一项差异；明细在 `actual_json.comparisons` | +| `DEFERRED_UNAVAILABLE` | 曾有候选预测，但阶段身份、旧路径或起点条件不满足；不计为匹配或算法不一致 | + +`actual_json` 含 `predictionCycleId`、`predictionInputHash`、`predictionEvidenceHash`、预测及观测时间、队列内容哈希和比较明细。引用的 cycleId 使用数据库实际返回的 ID，包含去重重放的情况。 + +后一周期仍为 `OBSERVATION_ONLY`，不生成新编码写入。原始预测已产生的虚拟编码保留在原周期。同一待下发区间持续出现多次时，使用最后一次成功提交的预测；已下发后持续观察不会反复计数。同一完整输入紧邻重放时保留原比较内容和哈希。 + +候选只保留上一成功采样的最多 500 条紧凑预测，不保留整批输入 JSON。采样失败、事实无效、任务消失、进程重启或时间间隔过大时不会跨越缺口配对。特殊交接阶段保持仅观察。无法配对的普通观测也可能没有 `DEFERRED_COMPARISON` 行，不能将“没有差异行”解释为通过。 + +## 现场更新 + +现场依然通过 IDEA 运行源码。与操作人员逐步进行，每步确认完成后再给下一步。 + +1. 只停止 IDEA 中的 `Ft546ShadowApplication`。不要停止 `RuoYiApplication`，不要使用 Stop All 或批量结束 Java。 +2. CMD 进入 `C:\project\FT546-WCS-SHANDOW`，执行 `git status --short`，保留现场配置和已有证据。若源码有重叠修改，先核对，不强制覆盖。 +3. 执行 `git fetch origin optimize/wcs`，再使用交付消息指定的准确提交执行 `git merge --ff-only <交付提交号>`。执行 `git log -1 --oneline` 核对。 +4. IDEA 刷新 Maven。沿用 Java 8、Maven 3.6.0 及现有离线仓库，编译影子模块和依赖。需要 CMD 编译时，在 Maven 已加入 PATH 的 CMD 中运行 `mvn -o -pl ruoyi-wcs-shadow -am test`；否则使用现场已有的 Maven 完整路径。 +5. 沿用现有数据库账号、只读源库连接、独立证据库、18091 本机端口、轮询间隔及容量上限。安全等待/避让点保持原配置，不凭推测新增。 +6. 本次启动使用新的运行记录，不复用旧 `runId`。人工 artifact/configuration 哈希按当前源码和现场配置重新计算；具体规则见 [9 月 8 日说明](ft546-shadow-20260908-phase-storage-repair.md#idea-中两个人工哈希字段)，不要把密码或配置快照提交到 Git。 +7. 只启动 `com.ruoyi.shadow.Ft546ShadowApplication`。在服务器本机浏览器访问 `http://127.0.0.1:18091/shadow/status`，核对新 buildId、启动检查全部 PASS、`READY`，间隔至少一个采样周期刷新后 `lastCompletedCycle` 应继续前进。`evidenceBacklog=0` 单独不能证明正在采集。 + +这次源码交付没有代替现场执行上述停止、更新、编译或启动操作。旧影子进程即使还在显示心跳，也不会自动获得修复。 + +## 后续评价与回退 + +下一轮需要分别统计新运行内的 `DEFERRED_MATCH`、`DEFERRED_DIFFERENCE`、`DEFERRED_UNAVAILABLE`，并按任务阶段及 predictionCycleId 检查独立样本，不能把全部观察周期作为通过次数。导出配对时同时包含观测周期和其引用的预测周期；原始输入继续通过本机 `/shadow/inputs/` 获取并验证哈希。 + +如果触发证据存储/事务保护，先保留现场错误和运行信息，只停止影子程序。需要回退时，使用保留的旧源码目录或单独检出的 `af8b5f1`；保留新影子证据，不删除或重建数据库，不对现场工作目录执行强制 reset。旧版本分析工具不认识新分类时，应使用新版本导出器解读新运行记录。 + +仍然缺少的正式接管依据：现场确认的安全等待/避让点、拥堵与长期 HOLD 的业务处置验证、特殊交接阶段状态机证据，以及本次更新后的连续运行配对数据。`NO_SAFE_YIELD_POINT` 没有被隐藏或强行改为放行。 + +## 本地验证 + +先用失败测试复现超时永久停采与缺失延迟比较，再验证修复。测试涵盖已结束读取恢复、不可取消读取仍停采、回滚失败、发布耗时预算、提交后登记预测、前后事务重放、车辆和阶段变化、旧路径不适用、缺口与地图变化、重复队列、SQL 证据存储兼容性。 + +现场回放只使用 Git 外的审计输入，不连接生产数据库。2026-09-09 的 11 份输入是独立样本，不伪造连续轨迹、历史交通状态或历史配对结果。测试使用与现场一致的 `ShadowVehicleAssignmentPolicy` 绑定车辆限制。 + +最终验证命令： + +```text +mvn -o -pl ruoyi-wcs-shadow -am -Dshadow.replay.audit-directory=<20260908审计目录> -Dshadow.replay.current-directory=<20260909审计目录> test +``` + +没有私有审计文件的现场运行普通 `mvn -o -pl ruoyi-wcs-shadow -am test` 即可；外部样本回放会明确跳过，不能声称现场也执行了这些回放。 + +2026-09-09 本地最终结果：`BUILD SUCCESS`；共 332 项，331 项通过、0 失败、0 错误、1 项因旧 300 条归档文件缺失跳过。WCS 核心 128 项、协议 10 项、影子模块 194 项（含上述 1 项跳过）。9 月 8 日回放 25 条记录、17 份不同输入；9 月 9 日回放 11 份独立输入，得到 8 个授予、5 个保持、16 个观察结果，原始哈希全部保持。这些是空交通状态下的独立回归结果，不是现场历史调度重建或正式接管验收成绩。 diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/evidence/DeferredShadowComparison.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/evidence/DeferredShadowComparison.java new file mode 100644 index 0000000..74c3cb0 --- /dev/null +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/evidence/DeferredShadowComparison.java @@ -0,0 +1,205 @@ +package com.ruoyi.shadow.evidence; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.ruoyi.ft546.protocol.Ft546EncodedCommand; +import com.ruoyi.shadow.facts.LegacyQueueProtocolProjector; +import com.ruoyi.shadow.facts.ShadowFactBatch; +import com.ruoyi.shadow.facts.ShadowLegacyQueueResult; +import com.ruoyi.shadow.facts.ShadowQueueDispatchEligibility; +import com.ruoyi.shadow.facts.ShadowQueueEndpoints; +import com.ruoyi.shadow.facts.ShadowQueueFact; + +import java.time.Duration; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; + +/** Compares adjacent committed polls. Queue observations are not PLC acknowledgements. */ +final class DeferredShadowComparison { + private static final int MAX_PENDING = 500; + private static final Duration MAX_INTERVAL = Duration.ofSeconds(30); + private final ObjectMapper mapper; + private final LegacyQueueProtocolProjector projector; + private final ShadowComparisonService comparison; + private Map pending = Collections.emptyMap(); + private Map lastComparisons = Collections.emptyMap(); + private String lastBatchKey; + private String lastMapsKey; + private Instant lastCapturedAt; + + DeferredShadowComparison(ObjectMapper mapper, LegacyQueueProtocolProjector projector, + ShadowComparisonService comparison) { + this.mapper = mapper; + this.projector = projector; + this.comparison = comparison; + } + + synchronized Map prepare(ShadowFactBatch batch) { + String key = batchKey(batch); + if (key.equals(lastBatchKey)) return lastComparisons; + if (lastCapturedAt == null || batch.getDatabaseCaptureStartedAt() == null + || batch.getDatabaseCapturedAt() == null + || !batch.getDatabaseCaptureStartedAt().isAfter(lastCapturedAt) + || Duration.between(lastCapturedAt, batch.getDatabaseCapturedAt()).compareTo(MAX_INTERVAL) > 0 + || !mapsKey(batch).equals(lastMapsKey)) return Collections.emptyMap(); + Map result = new LinkedHashMap(); + for (ShadowQueueFact queue : uniqueActiveQueues(batch).values()) { + Prediction prediction = pending.get(queue.getGuid()); + if (prediction == null || !Integer.valueOf(1).equals(queue.getIssued()) + || !Integer.valueOf(3).equals(queue.getExecutionStatus())) continue; + ShadowLegacyQueueResult legacy = projector.project(queue, batch); + String unavailable = null; + if (!prediction.phaseKey.equals(phaseKey(queue))) unavailable = "PHASE_IDENTITY_CHANGED"; + else if (legacy.getStatus() != ShadowLegacyQueueResult.Status.AVAILABLE) unavailable = legacy.getReasonCode(); + else if (legacy.getPathPoints().isEmpty() || !sameSite(prediction.sourceSite, legacy.getPathPoints().get(0))) + unavailable = "PHASE_SOURCE_POSITION_CHANGED"; + List compared = unavailable == null + ? comparison.compare(legacy, prediction.decision, prediction.command) + : Collections.emptyList(); + boolean match = compared.size() == 1 && compared.get(0).getCategory() == ShadowDifferenceCategory.MATCH; + String reason = unavailable != null ? "DEFERRED_UNAVAILABLE" : match ? "DEFERRED_MATCH" : "DEFERRED_DIFFERENCE"; + Map trace = new LinkedHashMap(); + trace.put("pairingBasis", "ADJACENT_COMMITTED_PENDING_TO_ISSUED_QUEUE"); + trace.put("comparisonScope", "OBSERVED_QUEUE_ONLY_NOT_PLC_ACKNOWLEDGEMENT"); + trace.put("predictionCycleId", prediction.cycleId); + trace.put("predictionInputHash", prediction.inputHash); + trace.put("predictionEvidenceHash", prediction.evidenceHash); + trace.put("predictionCapturedAt", prediction.capturedAt); + trace.put("predictionCompletedAt", prediction.completedAt); + trace.put("predictionDecision", prediction.decision); + trace.put("predictionQueueHash", prediction.queueHash); + trace.put("phaseIdentity", prediction.phaseKey); + trace.put("observedQueueHash", queue.getContentHash()); + trace.put("observedCapturedAt", batch.getDatabaseCapturedAt()); + trace.put("intervalMillis", Duration.between(prediction.capturedAt, batch.getDatabaseCapturedAt()).toMillis()); + trace.put("unavailableReason", unavailable); + trace.put("comparisons", compared); + result.put(queue.getGuid(), new ShadowDifference(ShadowDifferenceCategory.DEFERRED_COMPARISON, + reason, json(legacy), json(trace))); + } + return Collections.unmodifiableMap(result); + } + + /** Called only after the outer evidence transaction has committed successfully. */ + synchronized void committed(ShadowFactBatch batch, List evidence, + List saved, + Map comparisons) { + String key = batchKey(batch); + if (key.equals(lastBatchKey)) return; // Exact replay retains the original pairing and persisted references. + if (saved.size() != evidence.size()) throw new ShadowEvidenceException("SAVED_CYCLE_COUNT_MISMATCH"); + if (batch.getDatabaseCapturedAt() == null + || (lastCapturedAt != null && !batch.getDatabaseCapturedAt().isAfter(lastCapturedAt))) { + clear(); + return; + } + Map> byPhase = new LinkedHashMap>(); + for (int i = 0; i < evidence.size(); i++) { + ShadowCycleEvidence value = evidence.get(i); + String phase = json(Arrays.asList(value.getParentTaskId(), value.getPhase())); + byPhase.computeIfAbsent(phase, ignored -> new ArrayList()).add(i); + } + Map next = new LinkedHashMap(); + for (ShadowQueueFact queue : uniqueActiveQueues(batch).values()) { + if (next.size() >= MAX_PENDING) break; + if (ShadowQueueDispatchEligibility.observationReason(queue, batch) != null + || queue.getCreatedAt() == null || blank(queue.getTaskNo()) || blank(queue.getAgvId())) continue; + String phase = "1".equals(queue.getTaskType()) ? "EMPTY_TO_PICKUP" + : "2".equals(queue.getTaskType()) ? "LOADED_TO_PLACE" : "LEGACY_PHASE_" + trim(queue.getTaskType()); + List indices = byPhase.get(json(Arrays.asList(trim(queue.getTaskNo()), phase))); + if (indices == null || indices.size() != 1) continue; + int index = indices.get(0); + ShadowCycleEvidence value = evidence.get(index); + ShadowEvidenceRepository.SavedCycle stored = saved.get(index); + if (stored == null || stored.getCycleId() == null + || !value.getEvidenceHash().equals(stored.getEvidenceHash())) continue; + ShadowDecision.Type type = value.prediction().getType(); + if (type != ShadowDecision.Type.GRANTED && type != ShadowDecision.Type.HELD + && type != ShadowDecision.Type.YIELDED) continue; + next.put(queue.getGuid(), new Prediction(phaseKey(queue), queue.getContentHash(), + ShadowQueueEndpoints.dispatchStart(queue, batch), stored.getCycleId(), stored.getEvidenceHash(), value)); + } + pending = Collections.unmodifiableMap(next); + lastComparisons = comparisons; + lastBatchKey = key; + lastMapsKey = mapsKey(batch); + lastCapturedAt = batch.getDatabaseCapturedAt(); + } + + synchronized void clear() { + pending = Collections.emptyMap(); + lastComparisons = Collections.emptyMap(); + lastBatchKey = null; + lastMapsKey = null; + lastCapturedAt = null; + } + + private Map uniqueActiveQueues(ShadowFactBatch batch) { + Map unique = new LinkedHashMap(); + Map counts = new LinkedHashMap(); + for (ShadowQueueFact queue : batch.getSourceQueues()) { + if (!queue.isActive()) continue; + counts.merge(queue.getGuid(), 1, Integer::sum); + unique.put(queue.getGuid(), queue); + } + unique.entrySet().removeIf(entry -> counts.get(entry.getKey()) != 1); + return unique; + } + + private String phaseKey(ShadowQueueFact q) { + return json(Arrays.asList(q.getGuid(), q.getCreatedAt(), trim(q.getTaskNo()), trim(q.getTaskType()), + trim(q.getStartSite()), trim(q.getEndSite()), trim(q.getTheFloor()), trim(q.getAgvId()), + trim(q.getCargoPickSite()), trim(q.getCargoPlaceSite()), q.getPriority(), + trim(q.getDefault02()), trim(q.getDefault03()), trim(q.getDefault04()), trim(q.getDefault05()), + trim(q.getDefault06()), trim(q.getDefault07()), trim(q.getDefault08()))); + } + + private String batchKey(ShadowFactBatch batch) { + List queues = new ArrayList(); + batch.getSourceQueues().forEach(q -> queues.add(q.getContentHash())); + List vehicles = new ArrayList(); + batch.getVehicles().forEach(v -> vehicles.add(v.getContentHash())); + return ShadowEvidenceDigest.sha256(json(Arrays.asList(batch.getDatabaseCaptureStartedAt(), + batch.getDatabaseCapturedAt(), queues, vehicles, mapsKey(batch)))); + } + + private String mapsKey(ShadowFactBatch batch) { return ShadowEvidenceDigest.sha256(json(batch.getMaps())); } + private String json(Object value) { return ShadowEvidenceDigest.canonicalJson(mapper, value); } + private static boolean blank(String value) { return trim(value).isEmpty(); } + private static String trim(String value) { return value == null ? "" : value.trim(); } + private static boolean sameSite(String value, Integer site) { + try { return Integer.valueOf(value).equals(site); } + catch (NumberFormatException invalid) { return false; } + } + + private static final class Prediction { + final String phaseKey; + final String queueHash; + final String sourceSite; + final UUID cycleId; + final String evidenceHash; + final String inputHash; + final Instant capturedAt; + final Instant completedAt; + final ShadowDecision decision; + final Ft546EncodedCommand command; + + Prediction(String phaseKey, String queueHash, String sourceSite, UUID cycleId, + String evidenceHash, ShadowCycleEvidence evidence) { + this.phaseKey = phaseKey; + this.queueHash = queueHash; + this.sourceSite = sourceSite; + this.cycleId = cycleId; + this.evidenceHash = evidenceHash; + this.inputHash = evidence.getInputBatchHash(); + this.capturedAt = evidence.getDatabaseCapturedAt(); + this.completedAt = evidence.getCompletedAt(); + this.decision = evidence.prediction(); + this.command = evidence.command(); + } + } +} diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/evidence/ShadowCycleEvidence.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/evidence/ShadowCycleEvidence.java index e9cbcc0..db1afba 100644 --- a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/evidence/ShadowCycleEvidence.java +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/evidence/ShadowCycleEvidence.java @@ -44,6 +44,9 @@ public final class ShadowCycleEvidence { private final List encodedWrites; private final String encodedHash; private final List differences; + // Kept only for bounded, post-commit pairing. Not part of the persisted payload or digest. + private final ShadowDecision prediction; + private final Ft546EncodedCommand command; private ShadowCycleEvidence(UUID runId, ShadowFactBatch batch, ShadowLegacyQueueResult legacy, @@ -53,6 +56,8 @@ public final class ShadowCycleEvidence { Instant completedAt, ObjectMapper objectMapper) { this.cycleId = UUID.randomUUID(); + this.prediction = decision; + this.command = encoded; this.runId = Objects.requireNonNull(runId, "runId"); this.parentTaskId = decision.getParentTaskId(); this.phase = decision.getPhase(); @@ -152,6 +157,8 @@ public final class ShadowCycleEvidence { } public String getEncodedHash() { return encodedHash; } public List getDifferences() { return differences; } + ShadowDecision prediction() { return prediction; } + Ft546EncodedCommand command() { return command; } private static Map inputPayload( ShadowFactBatch batch, ShadowDecision decision) { diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/evidence/ShadowCycleEvidenceCollector.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/evidence/ShadowCycleEvidenceCollector.java index 02380b2..c9aa1d3 100644 --- a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/evidence/ShadowCycleEvidenceCollector.java +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/evidence/ShadowCycleEvidenceCollector.java @@ -14,6 +14,7 @@ import java.time.Instant; import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.Map; import java.util.Objects; import java.util.UUID; @@ -26,6 +27,7 @@ public final class ShadowCycleEvidenceCollector { private final ObjectMapper objectMapper; private final LegacyQueueProtocolProjector legacyProjector; private final ShadowComparisonService comparison; + private final DeferredShadowComparison deferred; public ShadowCycleEvidenceCollector(UUID runId, long routeVersion, ObjectMapper objectMapper) { @@ -49,11 +51,19 @@ public final class ShadowCycleEvidenceCollector { this.legacyProjector = Objects.requireNonNull(legacyProjector, "legacyProjector"); this.comparison = Objects.requireNonNull(comparison, "comparison"); + this.deferred = new DeferredShadowComparison(objectMapper, legacyProjector, comparison); } + /** Pure evidence preview; persistent callers use prepare() and acknowledge the outer commit. */ public List collect( ShadowFactBatch batch, DispatchCycleResult decisions, List encodedCommands, Instant completedAt) { + return prepare(batch, decisions, encodedCommands, completedAt).getEvidence(); + } + + public CollectedEvidence prepare( + ShadowFactBatch batch, DispatchCycleResult decisions, + List encodedCommands, Instant completedAt) { Objects.requireNonNull(batch, "batch"); Objects.requireNonNull(decisions, "decisions"); Objects.requireNonNull(encodedCommands, "encodedCommands"); @@ -63,6 +73,7 @@ public final class ShadowCycleEvidenceCollector { } List evidence = new ArrayList(); + Map deferredComparisons = deferred.prepare(batch); for (int index = 0; index < decisions.getGranted().size(); index++) { GrantedDispatch grant = decisions.getGranted().get(index); Ft546EncodedCommand encoded = encodedCommands.get(index); @@ -81,11 +92,39 @@ public final class ShadowCycleEvidenceCollector { if (!queue.isActive()) continue; String reason = com.ruoyi.shadow.facts.ShadowQueueDispatchEligibility.observationReason(queue, batch); if (reason != null) { - evidence.add(complete(batch, legacy(queue, batch), ShadowDecision.observation( - parent(queue), phase(queue), routeVersion, queue.getAgvId(), reason), null, completedAt)); + ShadowLegacyQueueResult legacy = legacy(queue, batch); + ShadowDecision observation = ShadowDecision.observation( + parent(queue), phase(queue), routeVersion, queue.getAgvId(), reason); + List differences = new ArrayList(comparison.compare(legacy, observation, null)); + ShadowDifference paired = deferredComparisons.get(queue.getGuid()); + if (paired != null) differences.add(paired); + evidence.add(ShadowCycleEvidence.complete(runId, batch, legacy, observation, null, + differences, completedAt, objectMapper)); } } - return Collections.unmodifiableList(evidence); + return new CollectedEvidence(batch, Collections.unmodifiableList(evidence), deferredComparisons); + } + + public void invalidatePendingComparisons() { deferred.clear(); } + + public final class CollectedEvidence { + private final ShadowFactBatch batch; + private final List evidence; + private final Map comparisons; + + private CollectedEvidence(ShadowFactBatch batch, List evidence, + Map comparisons) { + this.batch = batch; + this.evidence = evidence; + this.comparisons = comparisons; + } + + public List getEvidence() { return evidence; } + + /** The caller must wait until its outer evidence transaction has committed. */ + public void committed(List saved) { + deferred.committed(batch, evidence, saved, comparisons); + } } public List insufficient( diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/evidence/ShadowDifferenceCategory.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/evidence/ShadowDifferenceCategory.java index d0b4762..70ad7b1 100644 --- a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/evidence/ShadowDifferenceCategory.java +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/evidence/ShadowDifferenceCategory.java @@ -12,5 +12,6 @@ public enum ShadowDifferenceCategory { ALGORITHM_HOLD, INSUFFICIENT_FACTS, SOURCE_GAP, - OBSERVATION_ONLY + OBSERVATION_ONLY, + DEFERRED_COMPARISON } diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/SqlServerShadowFactReader.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/SqlServerShadowFactReader.java index 26207f5..d3cdc0e 100644 --- a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/SqlServerShadowFactReader.java +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/SqlServerShadowFactReader.java @@ -3,6 +3,7 @@ package com.ruoyi.shadow.facts; import org.springframework.dao.DataAccessException; import org.springframework.jdbc.core.ConnectionCallback; import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.transaction.TransactionException; import org.springframework.transaction.support.TransactionTemplate; import java.sql.PreparedStatement; @@ -84,6 +85,9 @@ public final class SqlServerShadowFactReader { throw failed(failure.getReasonCode(), failure); } catch (DataAccessException failure) { throw failed(classify(failure), failure); + } catch (TransactionException failure) { + // A failed commit/rollback cannot be treated as a completed query timeout. + throw failed("SOURCE_TRANSACTION_FAILED", failure); } catch (RuntimeException failure) { throw failed("SOURCE_READ_FAILED", failure); } finally { diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/kernel/Ft546ShadowCycle.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/kernel/Ft546ShadowCycle.java index de9542a..d2e8519 100644 --- a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/kernel/Ft546ShadowCycle.java +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/kernel/Ft546ShadowCycle.java @@ -18,6 +18,7 @@ import com.ruoyi.wcs.application.dispatch.DispatchSnapshot; import java.time.Instant; import java.time.Clock; import java.util.List; +import java.util.ArrayList; import java.util.Objects; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Function; @@ -110,14 +111,16 @@ public final class Ft546ShadowCycle { return degrade(health.getReasonCode(), clock.instant()); } try { + List afterCommit = new ArrayList(); progress.enter("TRANSACTION_BEGIN"); ShadowCycleResult result = evidenceRepository.inTransaction( () -> { - ShadowCycleResult saved = runAndPersist(batch, now, progress); + ShadowCycleResult saved = runAndPersist(batch, now, progress, afterCommit); progress.enter("TRANSACTION_COMMIT"); return saved; }); progress.enter("RECORDING"); + afterCommit.forEach(Runnable::run); commandGateway.rememberAll(result.getEncodedCommands()); Instant completedAt = clock.instant(); Long lag = batch.getDatabaseCapturedAt() == null ? null @@ -127,12 +130,13 @@ public final class Ft546ShadowCycle { statusReport.markReady(completedAt, lag); return result; } catch (RuntimeException evidenceFailure) { + invalidatePendingComparisons(); return degrade(failureReason(evidenceFailure), clock.instant()); } } private ShadowCycleResult runAndPersist(ShadowFactBatch batch, - Instant now, ShadowCycleDiagnostics progress) { + Instant now, ShadowCycleDiagnostics progress, List afterCommit) { progress.enter("VALIDATION"); ShadowFactValidation validation = validator.validate(batch, now); if (validation.getStatus() @@ -169,16 +173,18 @@ public final class Ft546ShadowCycle { } progress.enter("EVIDENCE_BUILD"); progress.dispatch(batch, snapshot, result.getDecisions()); - List evidence = evidenceCollector.collect(batch, + ShadowCycleEvidenceCollector.CollectedEvidence evidence = evidenceCollector.prepare(batch, result.getDecisions(), encoded, clock.instant()); progress.enter("EVIDENCE_SAVE"); - evidenceRepository.saveAll(evidence); + List saved = evidenceRepository.saveAll(evidence.getEvidence()); + afterCommit.add(() -> evidence.committed(saved)); return ShadowCycleResult.comparable(result.getDecisions(), result.getIntents(), encoded); } private ShadowCycleResult persistInsufficient( ShadowFactBatch batch, String reasonCode, ShadowCycleDiagnostics progress) { + invalidatePendingComparisons(); progress.enter("EVIDENCE_BUILD"); List evidence = evidenceCollector.insufficient(batch, reasonCode, clock.instant()); @@ -188,6 +194,7 @@ public final class Ft546ShadowCycle { } private ShadowCycleResult degrade(String reasonCode, Instant now) { + invalidatePendingComparisons(); degradedReason.compareAndSet(null, reasonCode); String stoppedReason = degradedReason.get(); statusReport.markDegraded(stoppedReason, now); @@ -209,6 +216,10 @@ public final class Ft546ShadowCycle { return "EVIDENCE_STORAGE_UNAVAILABLE"; } + public void invalidatePendingComparisons() { + if (evidenceCollector != null) evidenceCollector.invalidatePendingComparisons(); + } + private ShadowCycleResult runWithoutEvidence(ShadowFactBatch batch, Instant now, ShadowCycleDiagnostics progress) { progress.enter("VALIDATION"); diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/kernel/Ft546ShadowScheduler.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/kernel/Ft546ShadowScheduler.java index 0e62440..cd37bfd 100644 --- a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/kernel/Ft546ShadowScheduler.java +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/kernel/Ft546ShadowScheduler.java @@ -114,6 +114,7 @@ public final class Ft546ShadowScheduler implements AutoCloseable { } public boolean runOnce() { + long startedNanos = System.nanoTime(); Instant now = clock.instant(); if (permanentlyStopped.get() || now.isBefore(nextAttemptAt) || !inFlight.compareAndSet(false, true)) { @@ -121,6 +122,7 @@ public final class Ft546ShadowScheduler implements AutoCloseable { } ShadowCycleDiagnostics diagnostics = new ShadowCycleDiagnostics(); lastAttemptAt = now; + diagnostics.enter("RUNTIME_PUBLICATION"); if (!publishRuntimeState()) { inFlight.set(false); return false; @@ -130,9 +132,13 @@ public final class Ft546ShadowScheduler implements AutoCloseable { Future future = null; try { if (permanentlyStopped.get()) return false; - future = cycleExecutor.submit(() -> executeCycle(now, diagnostics)); - CycleAttempt attempt = future.get(cycleDeadline.toMillis(), - TimeUnit.MILLISECONDS); + Instant overallDeadline = now.plus(cycleDeadline); + long remainingNanos = Math.min(cycleDeadline.toNanos() - (System.nanoTime() - startedNanos), + Duration.between(clock.instant(), overallDeadline).toNanos()); + if (remainingNanos <= 0L) throw new TimeoutException("runtime publication exhausted cycle budget"); + diagnostics.enter("QUEUED"); + future = cycleExecutor.submit(() -> executeCycle(overallDeadline, diagnostics)); + CycleAttempt attempt = future.get(remainingNanos, TimeUnit.NANOSECONDS); outcome = attempt.result.getStatusCode(); statusReport.recordCycleDiagnostics(diagnostics.finish(outcome, null)); if ("DEGRADED".equals(attempt.result.getStatusCode())) { @@ -212,32 +218,37 @@ public final class Ft546ShadowScheduler implements AutoCloseable { } } - private CycleAttempt executeCycle(Instant startedAt, ShadowCycleDiagnostics diagnostics) { + private CycleAttempt executeCycle(Instant overallDeadline, ShadowCycleDiagnostics diagnostics) { cycleWorker = Thread.currentThread(); activeStage = CycleStage.SOURCE_READ; diagnostics.enter("SOURCE_READ"); - ShadowFactBatch batch = factReader.read(cursor, - startedAt.plus(sourceReadDeadline)); + Instant sourceDeadline = clock.instant().plus(sourceReadDeadline); + if (sourceDeadline.isAfter(overallDeadline)) sourceDeadline = overallDeadline; + ShadowFactBatch batch = factReader.read(cursor, sourceDeadline); activeStage = CycleStage.SHADOW_CYCLE; return new CycleAttempt(batch, shadowCycle.run(batch, clock.instant(), diagnostics)); } private boolean handleFailure(Throwable failure) { + shadowCycle.invalidatePendingComparisons(); if (failure instanceof ShadowFactReadException) { ShadowFactReadException readFailure = (ShadowFactReadException) failure; - String timeoutReason = sourceTimeoutReason(readFailure); - if (timeoutReason != null) { - return stopPermanently(timeoutReason, readFailure, - CycleStage.SOURCE_READ); + if ("SOURCE_TRANSACTION_FAILED".equals(readFailure.getReasonCode()) + || "SOURCE_CYCLE_ALREADY_IN_FLIGHT".equals(readFailure.getReasonCode())) { + return stopPermanently(readFailure.getReasonCode(), readFailure, CycleStage.SOURCE_READ); } + // ExecutionException is delivered only after the reader and its transaction have unwound. + // A watchdog timeout still stops permanently because cancellation does not prove termination. + String timeoutReason = sourceTimeoutReason(readFailure); + String reason = timeoutReason == null ? readFailure.getReasonCode() : timeoutReason; Instant failedAt = clock.instant(); - statusReport.markDegraded(readFailure.getReasonCode(), failedAt); - nextAttemptAt = failedAt.plus(readFailure.getRetryAfter()); + Duration backoff = readFailure.getRetryAfter().isZero() ? CRASH_BACKOFF : readFailure.getRetryAfter(); + statusReport.markDegraded(reason, failedAt); + nextAttemptAt = failedAt.plus(backoff); LOGGER.warn("FT546 shadow source read failed: {}; retry after {} ms", - readFailure.getReasonCode(), - readFailure.getRetryAfter().toMillis()); + reason, backoff.toMillis()); if (readFailure.getCause() != null) { LOGGER.warn("FT546 shadow source read root cause", readFailure.getCause()); @@ -278,6 +289,7 @@ public final class Ft546ShadowScheduler implements AutoCloseable { private boolean stopPermanently(String reasonCode, Exception failure, CycleStage stage) { + shadowCycle.invalidatePendingComparisons(); permanentlyStopped.set(true); nextAttemptAt = Instant.EPOCH; statusReport.markPermanentlyDegraded(reasonCode, clock.instant()); diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/startup/ShadowRuntimeIdentity.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/startup/ShadowRuntimeIdentity.java index 39966c5..c0132bf 100644 --- a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/startup/ShadowRuntimeIdentity.java +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/startup/ShadowRuntimeIdentity.java @@ -14,7 +14,7 @@ import java.util.TreeMap; /** Identifies this source revision and the nonsecret dispatch/storage settings, not a built artifact. */ public final class ShadowRuntimeIdentity { - public static final String BUILD_ID = "ft546-shadow-20260908-phase-storage-v2"; + public static final String BUILD_ID = "ft546-shadow-20260909-recovery-pairing-v3"; private ShadowRuntimeIdentity() { } public static Map describe(ShadowProperties properties, Environment env) { diff --git a/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/Ft546ShadowApplicationTest.java b/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/Ft546ShadowApplicationTest.java index cd519df..b2bdf8c 100644 --- a/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/Ft546ShadowApplicationTest.java +++ b/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/Ft546ShadowApplicationTest.java @@ -139,7 +139,7 @@ class Ft546ShadowApplicationTest { .andExpect(jsonPath("$.status").value("STARTING")) .andExpect(jsonPath("$.checkedAt").exists()) .andExpect(jsonPath("$.evidenceBacklog").value(0)) - .andExpect(jsonPath("$.runtimeIdentity.buildId").value("ft546-shadow-20260908-phase-storage-v2")) + .andExpect(jsonPath("$.runtimeIdentity.buildId").value("ft546-shadow-20260909-recovery-pairing-v3")) .andExpect(jsonPath("$.runtimeIdentity.effectiveSettingsSha256").exists()) .andExpect(jsonPath("$.runtimeIdentity.safePointConfiguration").value("UNCONFIGURED")) .andExpect(jsonPath("$.startupChecks.SHADOW_ENABLED") diff --git a/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/Ft546ShadowFaultIsolationTest.java b/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/Ft546ShadowFaultIsolationTest.java index b07cc09..606f27d 100644 --- a/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/Ft546ShadowFaultIsolationTest.java +++ b/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/Ft546ShadowFaultIsolationTest.java @@ -116,7 +116,7 @@ class Ft546ShadowFaultIsolationTest { } @Test - void retryAndPermanentFailurePublishDifferentCollectionStates() { + void completedSourceFailuresPublishRetryableCollectionState() { for (boolean timeout : new boolean[] {false, true}) { MutableClock clock = new MutableClock(NOW); AtomicInteger calls = new AtomicInteger(); @@ -137,7 +137,7 @@ class Ft546ShadowFaultIsolationTest { assertFalse(scheduler.runOnce()); Map failed = heartbeats.get(heartbeats.size() - 1); assertEquals("DEGRADED", failed.get("status")); - assertEquals(Boolean.valueOf(timeout), failed.get("collectionStopped")); + assertEquals(Boolean.FALSE, failed.get("collectionStopped")); assertEquals(null, failed.get("lastCompletedCycle")); clock.advance(Duration.ofMillis(100)); scheduler.runScheduledCycle(); @@ -172,7 +172,7 @@ class Ft546ShadowFaultIsolationTest { } @Test - void sourceQueryTimeoutPermanentlyStopsCollectionBeforeCycle() { + void completedSourceQueryTimeoutBacksOffWithoutRunningCycle() { MutableClock clock = new MutableClock(NOW); AtomicInteger sourceCalls = new AtomicInteger(); SqlServerShadowFactReader reader = reader(timeoutSource(sourceCalls), clock); @@ -197,17 +197,18 @@ class Ft546ShadowFaultIsolationTest { clock.advance(Duration.ofMinutes(1)); assertFalse(scheduler.runOnce()); - assertEquals(1, sourceCalls.get()); + assertEquals(2, sourceCalls.get()); assertEquals(0, evidence.transactionCalls.get()); assertEquals(Collections.emptyList(), gateway.getRecorded()); assertEquals(ShadowStartupReport.Status.DEGRADED, report.getStatus()); assertEquals("SOURCE_QUERY_TIMEOUT", report.getReasonCode()); - assertTrue(scheduler.isPermanentlyStopped()); + assertFalse(scheduler.isPermanentlyStopped()); + assertEquals(Duration.ofSeconds(2), scheduler.getRemainingBackoff()); assertTrue(logAppender.list.stream().anyMatch(event -> event.getFormattedMessage() - .contains("stage=SOURCE_READ")), - "source timeout log must identify the blocked stage"); + .contains("retry after")), + "completed source timeout must report the retry delay"); } finally { schedulerLogger.detachAppender(logAppender); logAppender.stop(); @@ -216,7 +217,7 @@ class Ft546ShadowFaultIsolationTest { } @Test - void sourceBatchDeadlinePermanentlyStopsCollection() { + void completedSourceBatchDeadlineBacksOff() { MutableClock clock = new MutableClock(NOW); AtomicInteger sourceCalls = new AtomicInteger(); SqlServerShadowFactReader.FactSource source = new FactSourceAdapter() { @@ -241,17 +242,18 @@ class Ft546ShadowFaultIsolationTest { clock.advance(Duration.ofMinutes(1)); assertFalse(scheduler.runOnce()); - assertEquals(1, sourceCalls.get()); + assertEquals(2, sourceCalls.get()); assertEquals("SOURCE_BATCH_DEADLINE_EXCEEDED", report.getReasonCode()); - assertTrue(scheduler.isPermanentlyStopped()); + assertFalse(scheduler.isPermanentlyStopped()); + assertEquals(Duration.ofSeconds(2), scheduler.getRemainingBackoff()); } finally { scheduler.close(); } } @Test - void wrappedSourceQueryTimeoutPermanentlyStopsCollection() { + void completedWrappedSourceQueryTimeoutBacksOff() { MutableClock clock = new MutableClock(NOW); AtomicInteger sourceCalls = new AtomicInteger(); SqlServerShadowFactReader.FactSource source = new FactSourceAdapter() { @@ -277,9 +279,10 @@ class Ft546ShadowFaultIsolationTest { clock.advance(Duration.ofMinutes(1)); assertFalse(scheduler.runOnce()); - assertEquals(1, sourceCalls.get()); + assertEquals(2, sourceCalls.get()); assertEquals("SOURCE_QUERY_TIMEOUT", report.getReasonCode()); - assertTrue(scheduler.isPermanentlyStopped()); + assertFalse(scheduler.isPermanentlyStopped()); + assertEquals(Duration.ofSeconds(2), scheduler.getRemainingBackoff()); } finally { scheduler.close(); } @@ -511,6 +514,39 @@ class Ft546ShadowFaultIsolationTest { Ft546ShadowFaultIsolationTest::isOutboundEndpointKey)); } + @Test + void sourceWorkerThatIgnoresCancellationIsNeverRetried() { + MutableClock clock = new MutableClock(NOW); + AtomicInteger calls = new AtomicInteger(); + CountDownLatch release = new CountDownLatch(1); + SqlServerShadowFactReader.FactSource source = new FactSourceAdapter() { + @Override public Instant readDatabaseClock(Instant deadline) { + calls.incrementAndGet(); + boolean interrupted = false; + while (release.getCount() != 0) { + try { release.await(); } + catch (InterruptedException ignored) { interrupted = true; } + } + if (interrupted) Thread.currentThread().interrupt(); + return clock.instant(); + } + }; + ShadowStartupReport report = report(); + try (Ft546ShadowScheduler scheduler = new Ft546ShadowScheduler(reader(source, clock), + evidenceCycle(report, new RecordingEvidenceRepository(), gateway(), + () -> ShadowEvidenceStorageHealth.Check.available(Long.MAX_VALUE)), + report, clock, Duration.ofSeconds(1))) { + assertFalse(scheduler.runOnce()); + assertEquals("SHADOW_CYCLE_TIMEOUT", report.getReasonCode()); + assertTrue(scheduler.isPermanentlyStopped()); + clock.advance(Duration.ofMinutes(1)); + assertFalse(scheduler.runOnce()); + assertEquals(1, calls.get()); + } finally { + release.countDown(); + } + } + @Test void longCycleDeadlineKeepsSourceReadDeadlineCappedAtFourSeconds() { MutableClock clock = new MutableClock(NOW); @@ -578,6 +614,93 @@ class Ft546ShadowFaultIsolationTest { assertEquals("", observed.get(1).getLastGuid()); } + @Test + void completedTimeoutRecoversOnlyAfterBackoffAndResetsFailureCount() { + MutableClock clock = new MutableClock(NOW); + AtomicInteger calls = new AtomicInteger(); + ShadowStartupReport report = report(); + SqlServerShadowFactReader.FactSource source = new FactSourceAdapter() { + @Override public Instant readDatabaseClock(Instant deadline) { + if (calls.incrementAndGet() == 1) throw new QueryTimeoutException("completed timeout"); + return clock.instant(); + } + }; + try (Ft546ShadowScheduler scheduler = new Ft546ShadowScheduler(reader(source, clock), + evidenceCycle(report, new RecordingEvidenceRepository(), gateway(), + () -> ShadowEvidenceStorageHealth.Check.available(Long.MAX_VALUE)), report, clock)) { + assertFalse(scheduler.runOnce()); + assertFalse(scheduler.runOnce()); + assertEquals(1, calls.get()); + clock.advance(Duration.ofSeconds(1)); + assertTrue(scheduler.runOnce()); + assertEquals(3, calls.get()); + assertEquals(ShadowStartupReport.Status.READY, report.getStatus()); + assertEquals(null, report.getReasonCode()); + assertFalse(scheduler.isPermanentlyStopped()); + } + } + + @Test + void failedSourceRollbackPermanentlyStopsEvenWhenOriginalCauseWasTimeout() { + MutableClock clock = new MutableClock(NOW); + AtomicInteger calls = new AtomicInteger(); + DriverManagerDataSource dataSource = new DriverManagerDataSource( + "jdbc:h2:mem:" + UUID.randomUUID(), "sa", ""); + DataSourceTransactionManager transactions = new DataSourceTransactionManager(dataSource) { + @Override protected void doRollback( + org.springframework.transaction.support.DefaultTransactionStatus status) { + super.doRollback(status); + throw new org.springframework.transaction.TransactionSystemException("rollback failed"); + } + }; + SqlServerShadowFactReader source = new SqlServerShadowFactReader(timeoutSource(calls), + new TransactionTemplate(transactions), clock); + ShadowStartupReport report = report(); + try (Ft546ShadowScheduler scheduler = new Ft546ShadowScheduler(source, + evidenceCycle(report, new RecordingEvidenceRepository(), gateway(), + () -> ShadowEvidenceStorageHealth.Check.available(Long.MAX_VALUE)), report, clock)) { + assertFalse(scheduler.runOnce()); + assertEquals("SOURCE_TRANSACTION_FAILED", report.getReasonCode()); + assertTrue(scheduler.isPermanentlyStopped()); + clock.advance(Duration.ofMinutes(1)); + assertFalse(scheduler.runOnce()); + assertEquals(1, calls.get()); + } + } + + @Test + void sourceBudgetStartsAfterRuntimePublicationAndIsCappedByWholeCycle() { + for (long publicationMillis : new long[] {3230L, 7000L, 11000L}) { + MutableClock clock = new MutableClock(NOW); + AtomicInteger publications = new AtomicInteger(); + AtomicReference deadlineSeen = new AtomicReference(); + SqlServerShadowFactReader.FactSource source = new FactSourceAdapter() { + @Override public Instant readDatabaseClock(Instant deadline) { + deadlineSeen.set(deadline); + return clock.instant(); + } + }; + ShadowStartupReport report = report(); + try (Ft546ShadowScheduler scheduler = new Ft546ShadowScheduler(reader(source, clock), + evidenceCycle(report, new RecordingEvidenceRepository(), gateway(), + () -> ShadowEvidenceStorageHealth.Check.available(Long.MAX_VALUE)), + report, clock, Duration.ofSeconds(10), state -> { + if (publications.incrementAndGet() == 1) clock.advance(Duration.ofMillis(publicationMillis)); + })) { + if (publicationMillis > 10000L) { + assertFalse(scheduler.runOnce()); + assertEquals(null, deadlineSeen.get()); + assertEquals("SHADOW_CYCLE_TIMEOUT", report.getReasonCode()); + } else { + assertTrue(scheduler.runOnce()); + assertEquals(NOW.plusMillis(Math.min(publicationMillis + 4000L, 10000L)), deadlineSeen.get()); + Map phases = (Map) report.getCycleDiagnostics().get("phaseMillis"); + assertTrue(phases.containsKey("RUNTIME_PUBLICATION")); + } + } + } + } + private static SqlServerShadowFactReader reader( SqlServerShadowFactReader.FactSource source, Clock clock) { DriverManagerDataSource dataSource = new DriverManagerDataSource( diff --git a/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/evidence/JdbcShadowEvidenceRepositoryTest.java b/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/evidence/JdbcShadowEvidenceRepositoryTest.java index d72f089..373a8d9 100644 --- a/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/evidence/JdbcShadowEvidenceRepositoryTest.java +++ b/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/evidence/JdbcShadowEvidenceRepositoryTest.java @@ -285,6 +285,33 @@ class JdbcShadowEvidenceRepositoryTest { assertEquals(1, count("shadow_cycle")); } + @Test + void deferredComparisonPersistsAlongsideObservationUsingExistingSchema() { + ShadowCycleEvidence prediction = matchingCycle(); + ShadowEvidenceRepository.SavedCycle pending = repository.save(prediction); + Map trace = new LinkedHashMap(); + trace.put("predictionCycleId", pending.getCycleId()); + trace.put("predictionEvidenceHash", pending.getEvidenceHash()); + trace.put("predictionInputHash", prediction.getInputBatchHash()); + String actual = ShadowEvidenceDigest.canonicalJson(objectMapper, trace); + ShadowCycleEvidence observation = ShadowCycleEvidence.complete(run.getRunId(), + ShadowFactBatch.create(NOW.plusSeconds(5), NOW.plusSeconds(5), + Collections.emptyList(), Collections.emptyList(), Collections.emptyList()), legacyFields(), + ShadowDecision.observation("PARENT-1", "LOADED_TO_PLACE", 3L, "VEH-1", "QUEUE_PHASE_NOT_PENDING"), + null, Arrays.asList(new ShadowDifference(ShadowDifferenceCategory.OBSERVATION_ONLY, + "QUEUE_PHASE_NOT_PENDING", null, null), new ShadowDifference(ShadowDifferenceCategory.DEFERRED_COMPARISON, + "DEFERRED_MATCH", "{}", actual)), NOW.plusSeconds(5), objectMapper); + repository.save(observation); + assertEquals(actual, jdbc.queryForObject("SELECT actual_json FROM shadow_difference WHERE cycle_id=? AND category=?", + String.class, observation.getCycleId(), "DEFERRED_COMPARISON")); + assertEquals(0, jdbc.queryForObject("SELECT COUNT(*) FROM shadow_encoded_write WHERE cycle_id=?", + Integer.class, observation.getCycleId())); + assertEquals(pending.getEvidenceHash(), repository.findCycle(run.getRunId(), prediction.getInputBatchHash(), + prediction.getParentTaskId(), prediction.getPhase(), prediction.getRouteVersion()).getEvidenceHash()); + assertEquals(Long.valueOf(1), repository.dailyDigest(LocalDate.of(2026, 8, 14), ZoneOffset.UTC) + .getCategoryCounts().get(ShadowDifferenceCategory.DEFERRED_COMPARISON)); + } + @Test void sameIdentityWithDifferentHashFailsClosed() { ShadowCycleEvidence first = matchingCycle(); diff --git a/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/evidence/ShadowDeferredComparisonTest.java b/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/evidence/ShadowDeferredComparisonTest.java new file mode 100644 index 0000000..027f705 --- /dev/null +++ b/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/evidence/ShadowDeferredComparisonTest.java @@ -0,0 +1,311 @@ +package com.ruoyi.shadow.evidence; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.ruoyi.ft546.protocol.*; +import com.ruoyi.shadow.facts.*; +import com.ruoyi.shadow.kernel.*; +import com.ruoyi.shadow.startup.ShadowStartupReport; +import com.ruoyi.wcs.application.ReadOnlyDispatchCycle; +import com.ruoyi.wcs.application.dispatch.DispatchCoordinator; +import com.ruoyi.wcs.domain.dispatch.DefaultVehicleAssignmentPolicy; +import com.ruoyi.wcs.domain.traffic.*; +import com.ruoyi.wcs.port.out.DispatchPolicy; +import org.junit.jupiter.api.Test; + +import java.time.*; +import java.util.*; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Supplier; + +import static org.junit.jupiter.api.Assertions.*; + +class ShadowDeferredComparisonTest { + private static final Instant NOW = Instant.parse("2026-09-09T01:00:00Z"); + private static final ObjectMapper JSON = new ObjectMapper().findAndRegisterModules(); + + @Test + void committedPendingPredictionPairsWithLaterIssuedPhaseWithoutChangingOldEvidence() throws Exception { + Harness h = new Harness(); + ShadowCycleEvidence pending = h.run(0, queue(0, null)); + String originalHash = pending.getEvidenceHash(); + assertEquals("GRANTED", pending.getDecision()); + assertEquals("SOURCE_GAP", pending.getDifferences().get(0).getCategory().name()); + ShadowCycleEvidence observed = h.run(5, queue(1, "70501,70502")); + ShadowDifference paired = deferred(observed); + assertEquals("DEFERRED_MATCH", paired.getReasonCode()); + JsonNode trace = JSON.readTree(paired.getActualJson()); + assertEquals(pending.getCycleId().toString(), trace.get("predictionCycleId").asText()); + assertEquals(pending.getInputBatchHash(), trace.get("predictionInputHash").asText()); + assertEquals(originalHash, trace.get("predictionEvidenceHash").asText()); + assertEquals("OBSERVATION_ONLY", observed.getDecision()); + assertTrue(observed.getEncodedWrites().isEmpty()); + assertEquals(originalHash, pending.getEvidenceHash()); + assertEquals(1, h.gateway.getRecorded().size()); + assertNull(deferredOrNull(h.run(10, queue(1, "70501,70502")))); + } + + @Test + void usesLastCommittedPredictionAndActualPersistedId() throws Exception { + Harness h = new Harness(); + h.run(0, queue(0, null)); + h.repository.returnedId = UUID.randomUUID(); + ShadowCycleEvidence last = h.run(5, queue(0, null)); + UUID persisted = h.repository.returnedId; + h.repository.returnedId = null; + JsonNode trace = JSON.readTree(deferred(h.run(10, queue(1, "70501,70502"))).getActualJson()); + assertEquals(persisted.toString(), trace.get("predictionCycleId").asText()); + assertEquals(last.getEvidenceHash(), trace.get("predictionEvidenceHash").asText()); + } + + @Test + void changedRouteIsADifferenceInsteadOfFalseMatch() { + Harness h = new Harness(); + h.run(0, queue(0, null)); + ShadowQueueFact changed = builder(1, "70501,70502").turnPath("70503").build(); + ShadowDifference difference = deferred(h.run(5, changed)); + assertEquals("DEFERRED_DIFFERENCE", difference.getReasonCode()); + assertTrue(difference.getActualJson().contains("ALGORITHM_ROUTE")); + } + + @Test + void neverMatchesChangedPhaseTaskIdentityVehicleOrSpecialHandoff() { + List changed = Arrays.asList( + builder(1, "70501,70502").taskType("1").build(), + builder(1, "70501,70502").taskNo("30002").build(), + builder(1, "70501,70502").createdAt(NOW.minusSeconds(1)).build(), + builder(1, "70501,70502").agvId("VEH-8").build(), + builder(1, "70501,70502").startSite("70503").build(), + builder(1, "70501,70502").default02("1").build(), + builder(1, "70501,70502").default08("2").build()); + for (ShadowQueueFact row : changed) { + Harness h = new Harness(); + h.run(0, queue(0, null)); + ShadowDifference result = deferredOrNull(h.run(5, row)); + assertTrue(result == null || "DEFERRED_UNAVAILABLE".equals(result.getReasonCode()), row.getContentHash()); + } + } + + @Test + void missingOrStaleLegacyPathRemainsUnavailable() { + for (String path : new String[] {null, "70502,70501"}) { + Harness h = new Harness(); + h.run(0, queue(0, null)); + ShadowDifference result = deferred(h.run(5, queue(1, path))); + assertEquals("DEFERRED_UNAVAILABLE", result.getReasonCode()); + } + } + + @Test + void noPredictionAcrossRestartMissingPollExpiredIntervalOrInvalidBatch() { + Harness fresh = new Harness(); + assertNull(deferredOrNull(fresh.run(5, queue(1, "70501,70502")))); + Harness expired = new Harness(); + expired.run(0, queue(0, null)); + assertNull(deferredOrNull(expired.run(120, queue(1, "70501,70502")))); + Harness gap = new Harness(); + gap.run(0, queue(0, null)); + gap.runBatch(gap.batch(5, Collections.emptyList(), true)); + assertNull(deferredOrNull(gap.run(10, queue(1, "70501,70502")))); + Harness invalid = new Harness(); + invalid.run(0, queue(0, null)); + invalid.runBatch(invalid.batch(5, Collections.singletonList(queue(0, null)), false)); + assertNull(deferredOrNull(invalid.run(10, queue(1, "70501,70502")))); + } + + @Test + void failedOuterCommitCannotCreateAPairablePrediction() { + Harness h = new Harness(); + h.repository.failCommit = true; + assertEquals("DEGRADED", h.runBatch(h.batch(0, Collections.singletonList(queue(0, null)), true)).getStatusCode()); + h.repository.failCommit = false; + // Reuse the collector deliberately: even an in-process reconstruction must not retain a rolled-back prediction. + h.cycle = h.newCycle(); + assertNull(deferredOrNull(h.run(5, queue(1, "70501,70502")))); + } + + @Test + void replayingSameObservedBatchKeepsIdenticalEvidenceHash() { + Harness h = new Harness(); + h.run(0, queue(0, null)); + ShadowCycleEvidence first = h.run(5, queue(1, "70501,70502")); + assertNotNull(deferred(first)); + ShadowCycleEvidence replay = h.run(5, queue(1, "70501,70502")); + assertEquals(first.getEvidenceHash(), replay.getEvidenceHash()); + } + + @Test + void sourceTimeoutBreaksPairingBeforeSuccessfulRecovery() { + Harness h = new Harness(); + AtomicReference facts = new AtomicReference( + h.batch(0, Collections.singletonList(queue(0, null)), true)); + java.util.concurrent.atomic.AtomicBoolean fail = new java.util.concurrent.atomic.AtomicBoolean(); + SqlServerShadowFactReader.FactSource source = new SqlServerShadowFactReader.FactSource() { + public Instant readDatabaseClock(Instant deadline) { + if (fail.get()) throw new org.springframework.dao.QueryTimeoutException("completed timeout"); + return facts.get().getDatabaseCapturedAt(); + } + public List readQueues(ShadowReadCursor cursor, Instant deadline) { return facts.get().getSourceQueues(); } + public List readVehicles(Instant deadline) { return facts.get().getVehicles(); } + public List readMaps(Instant deadline) { return facts.get().getMaps(); } + }; + org.springframework.jdbc.datasource.DriverManagerDataSource dataSource = + new org.springframework.jdbc.datasource.DriverManagerDataSource("jdbc:h2:mem:" + UUID.randomUUID(), "sa", ""); + SqlServerShadowFactReader reader = new SqlServerShadowFactReader(source, + new org.springframework.transaction.support.TransactionTemplate( + new org.springframework.jdbc.datasource.DataSourceTransactionManager(dataSource)), h.clock); + ShadowStartupReport status = ShadowStartupReport.starting(NOW, Collections.emptyMap()); + try (Ft546ShadowScheduler scheduler = new Ft546ShadowScheduler(reader, h.cycle, status, h.clock)) { + assertTrue(scheduler.runOnce()); + fail.set(true); + h.now.set(NOW.plusSeconds(5)); + assertFalse(scheduler.runOnce()); + fail.set(false); + h.now.set(NOW.plusSeconds(10)); + facts.set(h.batch(10, Collections.singletonList(queue(1, "70501,70502")), true)); + assertTrue(scheduler.runOnce()); + assertNull(deferredOrNull(h.repository.saved.get(h.repository.saved.size() - 1))); + } + } + + @Test + void changedMapAndDuplicateQueueObservationsCannotMatch() { + Harness h = new Harness(); + h.run(0, queue(0, null)); + ShadowFactBatch observed = h.batch(5, Collections.singletonList(queue(1, "70501,70502")), true); + ShadowFactBatch changedMap = ShadowFactBatch.create(observed.getDatabaseCaptureStartedAt(), + observed.getDatabaseCapturedAt(), observed.getSourceQueues(), observed.getVehicles(), + Arrays.asList(new ShadowMapFact("MAP-1", "70501", 0, 0, 1, 1), + new ShadowMapFact("MAP-2", "70502", 2, 0, 1, 1))); + List evidence = h.collector.collect(changedMap, + com.ruoyi.wcs.application.dispatch.DispatchCycleResult.committed(Collections.emptyList(), + Collections.emptyList(), Collections.emptyList(), Collections.emptyList()), + Collections.emptyList(), NOW.plusSeconds(5)); + assertNull(deferredOrNull(evidence.get(0))); + ShadowFactBatch duplicate = h.batch(5, Arrays.asList(queue(1, "70501,70502"), queue(1, "70501,70502")), true); + evidence = h.collector.collect(duplicate, + com.ruoyi.wcs.application.dispatch.DispatchCycleResult.committed(Collections.emptyList(), + Collections.emptyList(), Collections.emptyList(), Collections.emptyList()), + Collections.emptyList(), NOW.plusSeconds(5)); + for (ShadowCycleEvidence value : evidence) assertNull(deferredOrNull(value)); + } + + @Test + void preparingSameObservationBeforeCommitIsDeterministicAndDoesNotConsumePrediction() { + Harness h = new Harness(); + h.run(0, queue(0, null)); + ShadowFactBatch observed = h.batch(5, Collections.singletonList(queue(1, "70501,70502")), true); + com.ruoyi.wcs.application.dispatch.DispatchCycleResult noDispatch = + com.ruoyi.wcs.application.dispatch.DispatchCycleResult.committed(Collections.emptyList(), + Collections.emptyList(), Collections.emptyList(), Collections.emptyList()); + ShadowCycleEvidence first = h.collector.collect(observed, noDispatch, Collections.emptyList(), NOW.plusSeconds(5)).get(0); + ShadowCycleEvidence retry = h.collector.collect(observed, noDispatch, Collections.emptyList(), NOW.plusSeconds(6)).get(0); + assertNotNull(deferred(first)); + assertNotNull(deferred(retry)); + assertEquals(first.getEvidenceHash(), retry.getEvidenceHash()); + assertEquals(first.getEvidenceHash(), h.run(5, queue(1, "70501,70502")).getEvidenceHash()); + } + + private static ShadowDifference deferred(ShadowCycleEvidence value) { + ShadowDifference result = deferredOrNull(value); + assertNotNull(result, "same-stage pending->issued observation needs traced comparison evidence"); + return result; + } + + private static ShadowDifference deferredOrNull(ShadowCycleEvidence value) { + return value.getDifferences().stream().filter(d -> "DEFERRED_COMPARISON".equals(d.getCategory().name())) + .findFirst().orElse(null); + } + + private static ShadowQueueFact queue(int issued, String path) { return builder(issued, path).build(); } + private static ShadowQueueFact.Builder builder(int issued, String path) { + return ShadowQueueFact.builder("shadow_ro.v_wcs_srgv_queue", "ACTIVE") + .issued(issued).executionStatus(issued == 0 ? 1 : 3) + .taskNo("30001").taskType("2").priority(2L).theFloor("1") + .cargoPickSite("70501").cargoPlaceSite("70502").startSite("70501").endSite("70502") + .createdAt(NOW.minusSeconds(10)).agvId("VEH-9").pathInfo(path).default02("2"); + } + + private static final class Harness { + final Repository repository = new Repository(); + final ShadowCycleEvidenceCollector collector = new ShadowCycleEvidenceCollector(UUID.randomUUID(), 3L, JSON); + final AtomicReference now = new AtomicReference(NOW); + final Clock clock = new Clock() { + public ZoneId getZone() { return ZoneOffset.UTC; } + public Clock withZone(ZoneId zone) { return this; } + public Instant instant() { return now.get(); } + }; + final RecordingFt546CommandGateway gateway; + Ft546ShadowCycle cycle; + + Harness() { + Ft546DeviceProfile profile = new Ft546DeviceProfile(); + Map registers = new LinkedHashMap(); + registers.put("taskNo", "4000"); registers.put("taskType", "4002"); + registers.put("path", "4004"); registers.put("trigger", "4021"); + profile.setRegisters(registers); + Set capabilities = Collections.singleton("POINT_TO_POINT_COMMAND"); + profile.setCapabilities(capabilities); + gateway = new RecordingFt546CommandGateway(new Ft546SrgvCommandEncoder(), profile, capabilities); + cycle = newCycle(); + } + + Ft546ShadowCycle newCycle() { + RoutePlanner routes = new AStarRoutePlanner(); + DispatchCoordinator coordinator = new DispatchCoordinator( + new ShadowVehicleAssignmentPolicy(new DefaultVehicleAssignmentPolicy(routes, DispatchPolicy.NONE)), + new DeadlockGuard(new WaitForGraph(), routes), commit -> true, () -> UUID.randomUUID().toString()); + ShadowDispatchSnapshotAssembler assembler = new ShadowDispatchSnapshotAssembler("FT546", 3L, + topology -> new ShadowTrafficState(new ResourceLedger(topology), new WaitForGraph(), 0L, + Collections.emptyMap()), clock); + return new Ft546ShadowCycle(new ShadowFactBatchValidator(Duration.ofSeconds(4), Duration.ofSeconds(30)), + assembler::assemble, new ReadOnlyDispatchCycle(coordinator::dispatch, + new ShadowCommandIntentFactory()::create), gateway, + () -> ShadowEvidenceStorageHealth.Check.available(Long.MAX_VALUE), repository, collector, + ShadowStartupReport.starting(now.get(), Collections.emptyMap()), clock); + } + + ShadowFactBatch batch(long seconds, List queues, boolean vehicles) { + Instant at = NOW.plusSeconds(seconds); + return ShadowFactBatch.create(at.minusMillis(10), at, queues, + vehicles ? Collections.singletonList(ShadowVehicleFact.builder("VEH-GUID", "VEH-9") + .agvStatus("1").currentSite("70501").battery(80).workMode(5).alarm(0).status(1) + .errorCode(0).allowSend(1).currentFloor(1).capturedAt(at).build()) : Collections.emptyList(), + Arrays.asList(new ShadowMapFact("MAP-1", "70501", 0, 0, 1, 1), + new ShadowMapFact("MAP-2", "70502", 1, 0, 1, 1))); + } + + ShadowCycleResult runBatch(ShadowFactBatch batch) { + now.set(batch.getDatabaseCapturedAt()); + return cycle.run(batch, now.get()); + } + + ShadowCycleEvidence run(long seconds, ShadowQueueFact queue) { + assertEquals("COMPARABLE", runBatch(batch(seconds, Collections.singletonList(queue), true)).getStatusCode()); + return repository.saved.get(repository.saved.size() - 1); + } + } + + private static final class Repository implements ShadowEvidenceRepository { + final List saved = new ArrayList(); + boolean failCommit; + UUID returnedId; + public SavedCycle save(ShadowCycleEvidence value) { return saveAll(Collections.singletonList(value)).get(0); } + public List saveAll(List values) { + saved.addAll(values); + List result = new ArrayList(); + for (ShadowCycleEvidence value : values) result.add(new SavedCycle( + returnedId == null ? value.getCycleId() : returnedId, + value.getEvidenceHash(), value.getStatus(), value.getReasonCode())); + return result; + } + public T inTransaction(Supplier work) { + T result = work.get(); + if (failCommit) throw new ShadowEvidenceException("EVIDENCE_WRITE_FAILED"); + return result; + } + public SavedCycle findCycle(UUID run, String hash, String parent, String phase, long version) { return null; } + public ShadowEvidenceDigest dailyDigest(LocalDate date, ZoneId zone) { throw new UnsupportedOperationException(); } + public boolean isHealthy() { return true; } + } +} diff --git a/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/kernel/ShadowCurrentFieldReplayTest.java b/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/kernel/ShadowCurrentFieldReplayTest.java new file mode 100644 index 0000000..d4fb61f --- /dev/null +++ b/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/kernel/ShadowCurrentFieldReplayTest.java @@ -0,0 +1,91 @@ +package com.ruoyi.shadow.kernel; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.ruoyi.ft546.protocol.*; +import com.ruoyi.shadow.evidence.*; +import com.ruoyi.shadow.facts.*; +import com.ruoyi.wcs.application.ReadOnlyDispatchCycle; +import com.ruoyi.wcs.application.ShadowDispatchResult; +import com.ruoyi.wcs.application.dispatch.DispatchCoordinator; +import com.ruoyi.wcs.application.dispatch.DispatchSnapshot; +import com.ruoyi.wcs.domain.dispatch.DefaultVehicleAssignmentPolicy; +import com.ruoyi.wcs.domain.traffic.*; +import com.ruoyi.wcs.port.out.DispatchPolicy; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.EnabledIfSystemProperty; + +import java.nio.file.*; +import java.time.*; +import java.util.*; + +import static org.junit.jupiter.api.Assertions.*; + +/** Private representative samples are independent inputs, never a reconstructed continuous history. */ +@EnabledIfSystemProperty(named = "shadow.replay.current-directory", matches = ".+") +class ShadowCurrentFieldReplayTest { + @Test + void currentFieldInputsKeepHashesAndPredictionBoundaries() throws Exception { + ObjectMapper mapper = new ObjectMapper().findAndRegisterModules(); + int count = 0, grants = 0, holds = 0, observations = 0; + try (DirectoryStream paths = Files.newDirectoryStream( + Paths.get(System.getProperty("shadow.replay.current-directory")), "input-*.json")) { + for (Path path : paths) { + JsonNode export = mapper.readTree(path.toFile()); + String raw = export.get("capturedJson").asText(); + ShadowFactBatch batch = ShadowCapturedInputReplayTest.batch(mapper.readTree(raw)); + ShadowCycleEvidenceCollector collector = new ShadowCycleEvidenceCollector(UUID.randomUUID(), 1L, mapper); + ShadowCycleEvidence original = collector.insufficient(batch, "INDEPENDENT_INPUT_REPLAY", + batch.getDatabaseCapturedAt()).get(0); + assertEquals(raw, original.getInputJson()); + assertEquals(export.get("contentHash").asText(), original.getInputBatchHash()); + Clock clock = Clock.fixed(batch.getDatabaseCapturedAt(), ZoneOffset.UTC); + ShadowDispatchSnapshotAssembler assembler = new ShadowDispatchSnapshotAssembler("FT546", 1L, + topology -> new ShadowTrafficState(new ResourceLedger(topology), new WaitForGraph(), + 0L, Collections.emptyMap()), clock); + DispatchSnapshot snapshot = assembler.assemble(batch); + RoutePlanner routes = new AStarRoutePlanner(); + DispatchCoordinator coordinator = new DispatchCoordinator(new ShadowVehicleAssignmentPolicy( + new DefaultVehicleAssignmentPolicy(routes, DispatchPolicy.NONE)), + new DeadlockGuard(new WaitForGraph(), routes), commit -> true, () -> UUID.randomUUID().toString()); + ShadowDispatchResult result = new ReadOnlyDispatchCycle( + coordinator::dispatch, new ShadowCommandIntentFactory()::create).run(snapshot); + Ft546DeviceProfile profile = new Ft546DeviceProfile(); + Map registers = new LinkedHashMap(); + registers.put("taskNo", "4000"); registers.put("taskType", "4002"); + registers.put("path", "4004"); registers.put("trigger", "4021"); + profile.setRegisters(registers); + Set capabilities = Collections.singleton("POINT_TO_POINT_COMMAND"); + profile.setCapabilities(capabilities); + List encoded = new RecordingFt546CommandGateway( + new Ft546SrgvCommandEncoder(), profile, capabilities).previewAll(result.getIntents()); + for (ShadowCycleEvidence value : collector.collect(batch, result.getDecisions(), encoded, clock.instant())) { + assertEquals(original.getInputBatchHash(), value.getInputBatchHash()); + assertTrue(value.getDifferences().stream().noneMatch(d -> d.getCategory() == ShadowDifferenceCategory.DEFERRED_COMPARISON)); + if ("OBSERVATION_ONLY".equals(value.getDecision())) { + observations++; + assertTrue(value.getEncodedWrites().isEmpty()); + } + } + result.getDecisions().getGranted().forEach(grant -> { + ShadowQueueFact queue = batch.queue(grant.getStepId()); + assertNull(ShadowQueueDispatchEligibility.observationReason(queue, batch)); + if (queue.getAgvId() != null && !queue.getAgvId().trim().isEmpty()) + assertEquals(queue.getAgvId(), grant.getVehicleId()); + }); + if (export.get("cycleId").asText().equals("33d2f0b4-fa10-4908-a31c-c681bc6ded06")) { + ShadowQueueFact pending = batch.getQueues().stream().filter(q -> q.isActive() + && "60906".equals(q.getTaskNo())).findFirst().get(); + assertEquals("LEGACY_RESULT_UNAVAILABLE:INVALID_MAIN_PATH", + new LegacyQueueProtocolProjector().project(pending, batch).getReasonCode()); + } + grants += result.getDecisions().getGranted().size(); + holds += result.getDecisions().getHeld().size(); + count++; + } + } + assertEquals(11, count, "2026-09-09 representative export must be complete"); + System.out.println("CURRENT_FIELD_REPLAY independentInputs=" + count + " grants=" + grants + + " holds=" + holds + " observations=" + observations + " historicalPairingNotInferred=true"); + } +}