From 5d5668b04e16163bc3543acf65e45afa72dd0b9d Mon Sep 17 00:00:00 2001 From: zgw <13623675548@163.com> Date: Mon, 14 Sep 2026 09:57:57 +0800 Subject: [PATCH] fix(shadow): recover confirmed late evidence commits --- .../ft546-shadow-20260914-commit-recovery.md | 69 +++++ .../config/ShadowEvidenceConfiguration.java | 6 + .../ruoyi/shadow/kernel/Ft546ShadowCycle.java | 2 + .../shadow/kernel/Ft546ShadowScheduler.java | 145 ++++++++-- .../shadow/kernel/ShadowCycleDiagnostics.java | 11 +- .../shadow/startup/ShadowRuntimeIdentity.java | 2 +- .../shadow/startup/ShadowStartupReport.java | 21 +- .../shadow/Ft546ShadowApplicationTest.java | 4 +- .../shadow/ShadowCommitRecoveryTest.java | 260 ++++++++++++++++++ .../config/ShadowEvidenceTimeoutTest.java | 30 ++ .../kernel/ShadowCycleDiagnosticsTest.java | 22 ++ 11 files changed, 549 insertions(+), 23 deletions(-) create mode 100644 docs/runbooks/ft546-shadow-20260914-commit-recovery.md create mode 100644 ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/ShadowCommitRecoveryTest.java create mode 100644 ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/config/ShadowEvidenceTimeoutTest.java diff --git a/docs/runbooks/ft546-shadow-20260914-commit-recovery.md b/docs/runbooks/ft546-shadow-20260914-commit-recovery.md new file mode 100644 index 0000000..95ef488 --- /dev/null +++ b/docs/runbooks/ft546-shadow-20260914-commit-recovery.md @@ -0,0 +1,69 @@ +# FT546 影子程序：2026-09-14 证据提交超时修复 + +## 适用版本与范围 + +- 分支：`optimize/wcs` +- 修复前源码：`0d53f41` +- 新 buildId：`ft546-shadow-20260914-commit-recovery-v1` +- 修改范围：`ruoyi-wcs-shadow`。正式 WCS 调度、PLC/WMS 通信及生产数据库不作改动。 +- 无 SQL 迁移，无需清空影子库或更换 IDEA 连接配置；仍按现场源码方式编译启动。 + +本版本用于恢复持续影子采集，不代表批准新调度算法接管正式设备。 + +## 故障证据 + +9 月 14 日检查的旧 runId 为 `10ec02d4-5a64-4ea9-bfc4-6f03eca5089c`。 +最后成功周期为北京时间 9 月 11 日 02:31:03.399，此后心跳继续更新，采集却保持停止。 +超时诊断记录周期 10001 ms，其中前置心跳 4782 ms、来源读取 48 ms、证据提交 5154 ms； +工作线程当时位于 SQL Server JDBC 的提交/套接字读取路径。复查线程栈时，该工作线程已经空闲。 + +旧实现一旦超过周期看门狗预算，就取消 Future 并永久停采，即使原事务后来正常结束，也无法确认并恢复。 +现有证据不能确定数据库延迟由磁盘、锁、网络中的哪一项造成；本次修复不将这些推测当作已确认根因。 + +## 修复后的行为 + +1. 保留首次采集前的心跳检查。之后正常周期仅在完成后写一次心跳，减少证据库写入和前置耗时。 +2. 保留 10 秒周期预算。若超时发生在 `TRANSACTION_COMMIT` 或 `RECORDING`，保留原 Future，进入等待确认状态，不启动新批次、不重放旧事务，也不取消正在提交的 JDBC 调用。 +3. 后续调度检查原任务的终态。只有正常结束，包括事务清理及提交后回调完成，才确认迟到成功、推进原批次游标，并允许下一批采集；跨延迟区间的待比较数据会失效。 +4. 提交失败、确认丢失或等待期间心跳写入失败时仍停止采集。原任务持续未结束且等待超过 30 秒时，以 `SHADOW_COMMIT_COMPLETION_UNCONFIRMED` 停采。窗口按调度轮询检查，同时采用单调时间与墙上时间，避免时钟回拨导致无限等待。 +5. 超过等待窗口后，不再调度新批次，但原数据库事务仍可能随后提交。停止采集不等于已回滚，不能据此盲目重放原事务。 +6. 非提交阶段超时继续采用原有停止保护。人工关闭后，即使旧提交迟到完成，也不能将状态恢复为 READY。 + +仅影子证据 SQL Server 连接增加 `socketTimeout=15000` 毫秒及 `cancelQueryTimeout=1` 秒。 +现有语句超时 3 秒、连接池大小 2、连接获取等待 2 秒保持不变，来源只读连接配置保持不变。 +这些超时限制单次驱动等待，不保证整个周期或 JDBC 清理严格在 15 秒内返回。 + +## 如何判断是否仍在采集 + +在服务器浏览器打开 `http://127.0.0.1:18091/shadow/status`： + +- 正常：`status=READY`、`collectionStopped=false`、`awaitingCommitCompletion=false`；重复查看时 `lastCompletedCycle` 应持续推进。 +- 等待迟到提交：`awaitingCommitCompletion=true`，原因 `SHADOW_COMMIT_AWAITING_COMPLETION`。此时不启动新采集。 +- 停止采集：`collectionStopped=true`。进程存活或心跳时间更新不能作为持续采集的依据。 +- `lastCollectionFailure` 保留证据事务异常及可提取的 SQLState 等诊断；恢复后历史失败记录仍可能存在，应结合当前状态与时间判断。 + +迟到成功时日志包含 `late commit confirmed`。永久停采后每分钟输出一次 `collection remains stopped` 提醒。 + +## 本地验证 + +- 完整回归：核心 131 项、协议 10 项、影子 221 项，共 362 项；360 项通过，2 项因缺少另一组历史归档输入跳过，0 失败、0 错误。 +- 当前 11 份及占用场景 50 份现场输入回放通过。 +- 完整回归后补充了重复诊断快照的计时测试，单独运行该测试类 3 项全部通过；该补充之后未修改生产代码。 +- 使用真实 H2 事务注入提交延迟，验证迟到成功只提交一次、随后新批次继续、失败/确认丢失不重放、等待超限保护、关闭后不复活及心跳失败保护。 +- 使用 SQL Server JDBC 9.2.1.jre8 的属性解析验证有效超时配置，包括 URL 中无限等待值被覆盖的情形；此项未连接实际 SQL Server。 +- Java 8 `1.8.0_152`、Maven `3.6.0`，离线执行影子模块及其依赖的 `clean compile`，结果 `BUILD SUCCESS`。 +- 独立代码核验未发现阻止交付的问题，`git diff --check` 通过。 + +本地事务故障注入不等同于真实 SQL Server 网络故障试验。长期数据库抖动下的表现仍需重新采集现场运行记录。 + +## 现场更新与启动核对 + +现场影子源码目录为 `C:\project\FT546-WCS-SHANDOW`。更新前重新核实该目录版本、影子主类及 18091 监听进程， +只停止确认属于影子程序的旧进程。保留现场 `.gitignore`、IDEA 配置、`opencode.json` 与 `shadow-evidence`。 + +将影子分支快进到本次修复提交，使用现场 Java 8、Maven 3.6.0 及现有离线依赖库编译。 +核实新 buildId、Java class 主版本 52、正式 WCS/WMS 进程仍在运行，再由用户在 IDEA 启动 `Ft546ShadowApplication`。 +本说明描述更新流程，不单独证明现场已更新；以本次现场操作记录为准。 + +启动后确认新的 runId/buildId、全部启动检查 PASS，以及多个采集周期持续落库。 +后续上线评估按安全性、有效任务覆盖、载货运输及任务完成能力检查；新旧调度逐条一致不是验收门槛。 diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/config/ShadowEvidenceConfiguration.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/config/ShadowEvidenceConfiguration.java index 457cdab..dd8b84f 100644 --- a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/config/ShadowEvidenceConfiguration.java +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/config/ShadowEvidenceConfiguration.java @@ -31,6 +31,12 @@ public class ShadowEvidenceConfiguration { config.setMinimumIdle(0); config.setConnectionTimeout(2_000L); config.setInitializationFailTimeout(-1L); + if (config.getJdbcUrl() != null && config.getJdbcUrl().startsWith("jdbc:sqlserver:")) { + // Statement timeouts do not bound commit I/O. Allow a late acknowledgement, + // but never leave a socket or query cancellation waiting indefinitely. + config.addDataSourceProperty("socketTimeout", "15000"); + config.addDataSourceProperty("cancelQueryTimeout", "1"); + } return ShadowSourceConfiguration.lazyDataSource(config); } 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 d2e8519..15d1faa 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 @@ -131,6 +131,8 @@ public final class Ft546ShadowCycle { return result; } catch (RuntimeException evidenceFailure) { invalidatePendingComparisons(); + statusReport.recordCollectionFailure(failureReason(evidenceFailure), + "EVIDENCE_TRANSACTION", evidenceFailure, clock.instant(), null); return degrade(failureReason(evidenceFailure), clock.instant()); } } 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 575c6df..5bb575e 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 @@ -31,6 +31,7 @@ public final class Ft546ShadowScheduler implements AutoCloseable { private static final Duration CYCLE_DEADLINE = Duration.ofSeconds(10); private static final Duration SOURCE_READ_DEADLINE = Duration.ofSeconds(4); private static final Duration CRASH_BACKOFF = Duration.ofSeconds(1); + private static final Duration COMMIT_RECOVERY_WINDOW = Duration.ofSeconds(30); private static final String FINISH_QUEUE_VIEW = "shadow_ro.v_wcs_srgv_queue_finish"; @@ -52,6 +53,9 @@ public final class Ft546ShadowScheduler implements AutoCloseable { private volatile Instant nextAttemptAt = Instant.EPOCH; private volatile Thread cycleWorker; private long lastTimingLogNanos; + private long lastStoppedWarningNanos; + private boolean runtimeInitialized; + private volatile PendingCommit pendingCommit; public Ft546ShadowScheduler(SqlServerShadowFactReader factReader, Ft546ShadowCycle shadowCycle, @@ -107,6 +111,7 @@ public final class Ft546ShadowScheduler implements AutoCloseable { fixedDelayString = "${shadow.poll.fixed-delay-ms:5000}") public void runScheduledCycle() { if (permanentlyStopped.get() || clock.instant().isBefore(nextAttemptAt)) { + warnIfStopped(); publishRuntimeState(); } else { runOnce(); @@ -116,16 +121,26 @@ public final class Ft546ShadowScheduler implements AutoCloseable { public boolean runOnce() { long startedNanos = System.nanoTime(); Instant now = clock.instant(); - if (permanentlyStopped.get() || now.isBefore(nextAttemptAt) + if (closed.get() || permanentlyStopped.get() || now.isBefore(nextAttemptAt) || !inFlight.compareAndSet(false, true)) { return false; } + if (pendingCommit != null) { + try { + return reconcilePendingCommit(); + } finally { + try { publishRuntimeState(); } finally { inFlight.set(false); } + } + } ShadowCycleDiagnostics diagnostics = new ShadowCycleDiagnostics(); lastAttemptAt = now; - diagnostics.enter("RUNTIME_PUBLICATION"); - if (!publishRuntimeState()) { - inFlight.set(false); - return false; + if (!runtimeInitialized) { + diagnostics.enter("RUNTIME_PUBLICATION"); + if (!publishRuntimeState()) { + inFlight.set(false); + return false; + } + runtimeInitialized = true; } cycleWorker = null; String outcome = "FAILED"; @@ -154,6 +169,19 @@ public final class Ft546ShadowScheduler implements AutoCloseable { return true; } catch (TimeoutException timeout) { CycleStage timedOutStage = activeStage; + Map snapshot = diagnostics.snapshot("TIMEOUT", cycleWorker); + String phase = String.valueOf(snapshot.get("phase")); + if (future != null && ("TRANSACTION_COMMIT".equals(phase) || "RECORDING".equals(phase))) { + // Retain the future. cancel(true) loses the terminal result and can itself + // block inside JDBC cancellation. No new batch starts while this is pending. + pendingCommit = new PendingCommit(future, diagnostics, clock.instant()); + statusReport.recordCycleDiagnostics(snapshot); + statusReport.awaitCommitCompletion(clock.instant()); + statusReport.recordCollectionFailure("SHADOW_COMMIT_AWAITING_COMPLETION", phase, + timeout, clock.instant(), null); + LOGGER.warn("FT546 shadow commit exceeded cycle budget; awaiting worker completion; diagnostics={}", snapshot); + return false; + } statusReport.recordCycleDiagnostics(diagnostics.cancel("TIMEOUT", cycleWorker)); if (future != null) future.cancel(true); return stopPermanently("SHADOW_CYCLE_TIMEOUT", timeout, @@ -171,17 +199,73 @@ public final class Ft546ShadowScheduler implements AutoCloseable { } catch (java.util.concurrent.RejectedExecutionException shutdown) { return stopPermanently("SHADOW_EXECUTOR_STOPPED", shutdown, activeStage); } finally { - Map completed = diagnostics.finish(outcome, null); - statusReport.recordCycleDiagnostics(completed); - long logTime = System.nanoTime(); - if (!permanentlyStopped.get() && (lastTimingLogNanos == 0L - || logTime - lastTimingLogNanos >= TimeUnit.MINUTES.toNanos(1))) { - LOGGER.info("FT546 shadow cycle diagnostics: {}", completed); - lastTimingLogNanos = logTime; + if (pendingCommit == null) { + Map completed = diagnostics.finish(outcome, null); + statusReport.recordCycleDiagnostics(completed); + long logTime = System.nanoTime(); + if (!permanentlyStopped.get() && (lastTimingLogNanos == 0L + || logTime - lastTimingLogNanos >= TimeUnit.MINUTES.toNanos(1))) { + LOGGER.info("FT546 shadow cycle diagnostics: {}", completed); + lastTimingLogNanos = logTime; + } + activeStage = CycleStage.IDLE; } - inFlight.set(false); + try { publishRuntimeState(); } finally { inFlight.set(false); } + } + } + + private boolean reconcilePendingCommit() { + PendingCommit pending = pendingCommit; + if (!pending.future.isDone()) { + if (pending.expired(clock.instant())) { + statusReport.recordCycleDiagnostics(pending.diagnostics.cancel("COMMIT_UNCONFIRMED", cycleWorker)); + return stopPermanently("SHADOW_COMMIT_COMPLETION_UNCONFIRMED", + new TimeoutException("worker did not finish within commit recovery window"), CycleStage.SHADOW_CYCLE); + } + statusReport.recordCycleDiagnostics(pending.diagnostics.snapshot("AWAITING_COMMIT", cycleWorker)); + return false; + } + try { + CycleAttempt attempt = pending.future.get(); + pendingCommit = null; activeStage = CycleStage.IDLE; - publishRuntimeState(); + if (closed.get() || permanentlyStopped.get()) return false; + if ("DEGRADED".equals(attempt.result.getStatusCode())) { + permanentlyStopped.set(true); + statusReport.markPermanentlyDegraded(attempt.result.getReasonCode(), clock.instant()); + statusReport.recordCycleDiagnostics(pending.diagnostics.finish("COMMIT_FAILED", null)); + LOGGER.error("FT546 shadow late commit did not complete successfully; collection stopped: {}", + attempt.result.getReasonCode()); + return false; + } + // The future has completed normally, including JDBC cleanup and after-commit + // callbacks. Do not repeat the transaction or compare across this delayed gap. + shadowCycle.invalidatePendingComparisons(); + cursor = advanceFinishCursor(cursor, attempt.batch); + Long lag = attempt.batch.getDatabaseCapturedAt() == null ? null : Long.valueOf(Math.max(0L, + Duration.between(attempt.batch.getDatabaseCapturedAt(), attempt.completedAt).toMillis())); + statusReport.confirmCommitCompletion(attempt.completedAt, lag); + statusReport.recordCycleDiagnostics(pending.diagnostics.finish("COMMIT_CONFIRMED", null)); + nextAttemptAt = Instant.EPOCH; + LOGGER.info("FT546 shadow late commit confirmed; collection can resume from the next batch; completedAt={}", + attempt.completedAt); + return true; + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + return stopPermanently("SHADOW_CYCLE_INTERRUPTED", interrupted, CycleStage.SHADOW_CYCLE); + } catch (ExecutionException | java.util.concurrent.CancellationException failure) { + statusReport.recordCycleDiagnostics(pending.diagnostics.finish("COMMIT_FAILED", null)); + return stopPermanently("SHADOW_COMMIT_COMPLETION_FAILED", failure, CycleStage.SHADOW_CYCLE); + } + } + + private void warnIfStopped() { + if (!permanentlyStopped.get() || closed.get()) return; + long now = System.nanoTime(); + if (lastStoppedWarningNanos == 0L || now - lastStoppedWarningNanos >= TimeUnit.MINUTES.toNanos(1)) { + LOGGER.warn("FT546 shadow collection remains stopped; reason={}; lastCompletedCycle={}; heartbeat is not collection", + statusReport.getReasonCode(), statusReport.getLastCompletedCycle()); + lastStoppedWarningNanos = now; } } @@ -199,7 +283,13 @@ public final class Ft546ShadowScheduler implements AutoCloseable { if (!closed.compareAndSet(false, true)) return; permanentlyStopped.set(true); statusReport.markStopped(clock.instant()); - cycleExecutor.shutdownNow(); + PendingCommit pending = pendingCommit; + if (pending != null) { + pending.diagnostics.cancel("STOPPED", cycleWorker); + cycleExecutor.shutdown(); + } else { + cycleExecutor.shutdownNow(); + } publishRuntimeState(); } @@ -226,8 +316,8 @@ public final class Ft546ShadowScheduler implements AutoCloseable { 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)); + ShadowCycleResult result = shadowCycle.run(batch, clock.instant(), diagnostics); + return new CycleAttempt(batch, result, clock.instant()); } private boolean handleFailure(Throwable failure) { @@ -353,10 +443,29 @@ public final class Ft546ShadowScheduler implements AutoCloseable { private static final class CycleAttempt { private final ShadowFactBatch batch; private final ShadowCycleResult result; + private final Instant completedAt; - private CycleAttempt(ShadowFactBatch batch, ShadowCycleResult result) { + private CycleAttempt(ShadowFactBatch batch, ShadowCycleResult result, Instant completedAt) { this.batch = batch; this.result = result; + this.completedAt = completedAt; + } + } + + private static final class PendingCommit { + private final Future future; + private final ShadowCycleDiagnostics diagnostics; + private final Instant deadline; + private final long startedNanos = System.nanoTime(); + + private PendingCommit(Future future, ShadowCycleDiagnostics diagnostics, Instant startedAt) { + this.future = future; + this.diagnostics = diagnostics; + this.deadline = startedAt.plus(COMMIT_RECOVERY_WINDOW); + } + + private boolean expired(Instant now) { + return !now.isBefore(deadline) || System.nanoTime() - startedNanos >= COMMIT_RECOVERY_WINDOW.toNanos(); } } } diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/kernel/ShadowCycleDiagnostics.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/kernel/ShadowCycleDiagnostics.java index 5c86557..9c86e08 100644 --- a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/kernel/ShadowCycleDiagnostics.java +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/kernel/ShadowCycleDiagnostics.java @@ -77,9 +77,17 @@ final class ShadowCycleDiagnostics { } synchronized Map finish(String outcome, Thread worker) { + if (completed != null) return completed; + completed = snapshot(outcome, worker); + return completed; + } + + /** Observe an over-budget commit without cancelling its completion bookkeeping. */ + synchronized Map snapshot(String outcome, Thread worker) { if (completed != null) return completed; long now = ticker.getAsLong(); accumulate(now); + phaseStarted = now; Map snapshot = new LinkedHashMap(); snapshot.put("outcome", outcome); snapshot.put("phase", phase); @@ -95,8 +103,7 @@ final class ShadowCycleDiagnostics { for (int i = 0; i < Math.min(16, frames.length); i++) stack.add(frames[i].toString()); snapshot.put("workerStack", Collections.unmodifiableList(stack)); } - completed = Collections.unmodifiableMap(snapshot); - return completed; + return Collections.unmodifiableMap(snapshot); } synchronized Map cancel(String outcome, Thread worker) { 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 04c5dbf..6ca14c9 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-20260910-source-recovery-v1"; + public static final String BUILD_ID = "ft546-shadow-20260914-commit-recovery-v1"; private ShadowRuntimeIdentity() { } public static Map describe(ShadowProperties properties, Environment env) { diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/startup/ShadowStartupReport.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/startup/ShadowStartupReport.java index bea125b..24dd94b 100644 --- a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/startup/ShadowStartupReport.java +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/startup/ShadowStartupReport.java @@ -15,6 +15,7 @@ public final class ShadowStartupReport { private volatile long evidenceBacklog; private volatile String reasonCode; private volatile String permanentDegradedReason; + private volatile boolean awaitingCommitCompletion; private final Map startupChecks; private volatile Map cycleDiagnostics = Collections.emptyMap(); private volatile Map runtimeIdentity = Collections.emptyMap(); @@ -50,6 +51,21 @@ public final class ShadowStartupReport { public Map getCycleDiagnostics() { return cycleDiagnostics; } public Map getRuntimeIdentity() { return runtimeIdentity; } public Map getLastCollectionFailure() { return lastCollectionFailure; } + public boolean isCollectionStopped() { + return permanentDegradedReason != null || status == Status.STOPPED; + } + public boolean isAwaitingCommitCompletion() { return awaitingCommitCompletion; } + + public synchronized void awaitCommitCompletion(Instant checkedAt) { + if (isCollectionStopped()) return; + awaitingCommitCompletion = true; + markDegraded("SHADOW_COMMIT_AWAITING_COMPLETION", checkedAt); + } + + public synchronized void confirmCommitCompletion(Instant completedAt, Long lagMillis) { + awaitingCommitCompletion = false; + markReady(completedAt, lagMillis); + } public void recordCollectionFailure(String reason, String stage, Throwable failure, Instant occurredAt, Instant nextAttemptAt) { lastCollectionFailure = ShadowCollectionFailure.describe(reason, stage, failure, occurredAt, nextAttemptAt); @@ -64,7 +80,7 @@ public final class ShadowStartupReport { public synchronized void markReady(Instant completedAt, Long sourceLagMillis) { - if (permanentDegradedReason != null || status == Status.STOPPED) { + if (permanentDegradedReason != null || status == Status.STOPPED || awaitingCommitCompletion) { return; } this.status = Status.READY; @@ -95,10 +111,12 @@ public final class ShadowStartupReport { } this.status = Status.DEGRADED; this.reasonCode = permanentDegradedReason; + awaitingCommitCompletion = false; } public synchronized void markStopped(Instant stoppedAt) { status = Status.STOPPED; + awaitingCommitCompletion = false; checkedAt = stoppedAt; if (reasonCode == null) reasonCode = "SHADOW_APPLICATION_STOPPED"; } @@ -116,6 +134,7 @@ public final class ShadowStartupReport { state.put("reasonCode", reasonCode); state.put("collectionStopped", collectionStopped || status == Status.STOPPED || permanentDegradedReason != null); + state.put("awaitingCommitCompletion", awaitingCommitCompletion); state.put("sourceLagMillis", sourceLagMillis); state.put("cycleDiagnostics", cycleDiagnostics); state.put("runtimeIdentity", runtimeIdentity); 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 5a482bf..4a6b676 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,9 @@ class Ft546ShadowApplicationTest { .andExpect(jsonPath("$.status").value("STARTING")) .andExpect(jsonPath("$.checkedAt").exists()) .andExpect(jsonPath("$.evidenceBacklog").value(0)) - .andExpect(jsonPath("$.runtimeIdentity.buildId").value("ft546-shadow-20260910-source-recovery-v1")) + .andExpect(jsonPath("$.runtimeIdentity.buildId").value("ft546-shadow-20260914-commit-recovery-v1")) + .andExpect(jsonPath("$.collectionStopped").value(false)) + .andExpect(jsonPath("$.awaitingCommitCompletion").value(false)) .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/ShadowCommitRecoveryTest.java b/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/ShadowCommitRecoveryTest.java new file mode 100644 index 0000000..0bf5721 --- /dev/null +++ b/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/ShadowCommitRecoveryTest.java @@ -0,0 +1,260 @@ +package com.ruoyi.shadow; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.ruoyi.ft546.protocol.Ft546DeviceProfile; +import com.ruoyi.ft546.protocol.Ft546SrgvCommandEncoder; +import com.ruoyi.ft546.protocol.Ft546SrgvCommandIntent; +import com.ruoyi.shadow.evidence.*; +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.DispatchCycleResult; +import org.junit.jupiter.api.Test; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.DataSourceTransactionManager; +import org.springframework.jdbc.datasource.DriverManagerDataSource; +import org.springframework.transaction.TransactionSystemException; +import org.springframework.transaction.support.DefaultTransactionStatus; +import org.springframework.transaction.support.TransactionTemplate; + +import java.time.*; +import java.util.*; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Supplier; + +import static org.junit.jupiter.api.Assertions.*; + +class ShadowCommitRecoveryTest { + private static final Instant NOW = Instant.parse("2026-09-14T02:00:00Z"); + + @Test + void confirmedLateCommitResumesWithoutDuplicatingTheTransaction() throws Exception { + try (Fixture f = new Fixture(CommitResult.SUCCESS)) { + assertFalse(f.scheduler.runOnce()); + assertTrue(f.repository.entered.await(0, TimeUnit.MILLISECONDS)); + assertFalse(f.scheduler.isPermanentlyStopped(), "a pending commit must retain its result"); + assertTrue(json(f.report).path("awaitingCommitCompletion").asBoolean()); + assertFalse(json(f.report).path("collectionStopped").asBoolean()); + assertEquals(ShadowStartupReport.Status.DEGRADED, f.report.getStatus()); + assertEquals(0, f.repository.rows(), "uncommitted evidence must remain invisible"); + for (int i = 0; i < 3; i++) assertFalse(f.scheduler.runOnce()); + assertEquals(1, f.batches.get(), "waiting must not start another source read"); + + f.repository.release.countDown(); + assertTrue(f.repository.finished.await(2, TimeUnit.SECONDS)); + awaitResolution(f); + assertEquals(1, f.repository.rows(), "the late transaction must not be replayed"); + assertEquals(1, f.batches.get()); + assertEquals(ShadowStartupReport.Status.READY, f.report.getStatus()); + assertFalse(json(f.report).path("awaitingCommitCompletion").asBoolean()); + assertFalse(f.report.getLastCollectionFailure().isEmpty(), "keep timeout history after recovery"); + assertTrue(f.scheduler.runOnce()); + assertEquals(2, f.batches.get()); + assertEquals(2, f.repository.rows()); + } + } + + @Test + void unconfirmedCommitOutcomeNeverTriggersAnotherBatch() throws Exception { + for (CommitResult outcome : Arrays.asList(CommitResult.ROLLBACK, CommitResult.ACK_LOST)) { + try (Fixture f = new Fixture(outcome)) { + assertFalse(f.scheduler.runOnce()); + assertFalse(f.scheduler.isPermanentlyStopped()); + f.repository.release.countDown(); + assertTrue(f.repository.finished.await(2, TimeUnit.SECONDS)); + awaitResolution(f); + assertTrue(f.scheduler.isPermanentlyStopped()); + assertEquals("EVIDENCE_STORAGE_UNAVAILABLE", f.report.getReasonCode()); + assertTrue(json(f.report).path("collectionStopped").asBoolean()); + assertEquals(outcome == CommitResult.ACK_LOST ? 1 : 0, f.repository.rows()); + assertFalse(f.scheduler.runOnce()); + assertEquals(1, f.batches.get(), "a lost commit acknowledgement is not permission to retry"); + } + } + } + + @Test + void unfinishedCommitStopsAfterTheRecoveryWindowAndCannotRevive() throws Exception { + try (Fixture f = new Fixture(CommitResult.SUCCESS)) { + assertFalse(f.scheduler.runOnce()); + assertFalse(f.scheduler.isPermanentlyStopped()); + f.clock.advance(Duration.ofSeconds(31)); + assertFalse(f.scheduler.runOnce()); + assertTrue(f.scheduler.isPermanentlyStopped()); + assertEquals("SHADOW_COMMIT_COMPLETION_UNCONFIRMED", f.report.getReasonCode()); + f.repository.release.countDown(); + assertTrue(f.repository.finished.await(2, TimeUnit.SECONDS)); + f.report.markReady(f.clock.instant(), 0L); + assertEquals(ShadowStartupReport.Status.DEGRADED, f.report.getStatus()); + assertFalse(f.scheduler.runOnce()); + assertEquals(1, f.batches.get()); + } + } + + @Test + void closingWhileCommitIsPendingDoesNotPublishReadyLater() throws Exception { + try (Fixture f = new Fixture(CommitResult.SUCCESS)) { + assertFalse(f.scheduler.runOnce()); + assertTrue(json(f.report).path("awaitingCommitCompletion").asBoolean()); + f.scheduler.close(); + f.repository.release.countDown(); + assertTrue(f.repository.finished.await(2, TimeUnit.SECONDS)); + f.report.markReady(NOW.plusSeconds(10), 0L); + assertEquals(ShadowStartupReport.Status.STOPPED, f.report.getStatus()); + assertTrue(json(f.report).path("collectionStopped").asBoolean()); + assertFalse(f.scheduler.runOnce()); + assertEquals(1, f.batches.get()); + } + } + + @Test + void heartbeatFailureDuringPendingCommitPreventsLaterRecovery() throws Exception { + try (Fixture f = new Fixture(CommitResult.SUCCESS)) { + f.failPendingHeartbeat = true; + assertFalse(f.scheduler.runOnce()); + assertTrue(f.scheduler.isPermanentlyStopped()); + assertEquals("EVIDENCE_RUNTIME_WRITE_FAILED", f.report.getReasonCode()); + f.repository.release.countDown(); + assertTrue(f.repository.finished.await(2, TimeUnit.SECONDS)); + assertFalse(f.scheduler.runOnce()); + assertEquals(1, f.batches.get()); + } + } + + @Test + void healthyPollsPublishOnceAfterEachCycleFollowingTheStartupCheck() { + try (Fixture f = new Fixture(CommitResult.SUCCESS)) { + f.repository.release.countDown(); + assertTrue(f.scheduler.runOnce()); + assertTrue(f.scheduler.runOnce()); + assertEquals(3, f.heartbeats.size(), "one initial check plus one completion heartbeat per poll"); + assertEquals("STARTING", f.heartbeats.get(0).get("status")); + assertEquals("READY", f.heartbeats.get(2).get("status")); + assertEquals(2, f.repository.rows()); + } + } + + private static com.fasterxml.jackson.databind.JsonNode json(Object value) { + return new ObjectMapper().findAndRegisterModules().valueToTree(value); + } + + private static void awaitResolution(Fixture fixture) throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2); + while (System.nanoTime() < deadline) { + if (fixture.scheduler.runOnce() || fixture.scheduler.isPermanentlyStopped()) return; + Thread.sleep(5L); + } + fail("the completed worker was not reconciled"); + } + + private enum CommitResult { SUCCESS, ROLLBACK, ACK_LOST } + + private static final class Fixture implements AutoCloseable { + final MutableClock clock = new MutableClock(); + final AtomicInteger batches = new AtomicInteger(); + final ShadowStartupReport report = ShadowStartupReport.starting(NOW, Collections.emptyMap()); + final List> heartbeats = new ArrayList>(); + final CommitRepository repository; + final Ft546ShadowScheduler scheduler; + boolean failPendingHeartbeat; + + Fixture(CommitResult result) { + repository = new CommitRepository(result); + DriverManagerDataSource source = new DriverManagerDataSource( + "jdbc:h2:mem:" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1", "sa", ""); + SqlServerShadowFactReader reader = new SqlServerShadowFactReader(new SqlServerShadowFactReader.FactSource() { + public Instant readDatabaseClock(Instant deadline) { return clock.instant(); } + public List readQueues(ShadowReadCursor cursor, Instant deadline) { + batches.incrementAndGet(); return Collections.emptyList(); + } + public List readVehicles(Instant deadline) { return Collections.emptyList(); } + public List readMaps(Instant deadline) { return Collections.emptyList(); } + }, new TransactionTemplate(new DataSourceTransactionManager(source)), clock); + 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); + RecordingFt546CommandGateway gateway = new RecordingFt546CommandGateway( + new Ft546SrgvCommandEncoder(), profile, capabilities); + Ft546ShadowCycle cycle = new Ft546ShadowCycle( + new ShadowFactBatchValidator(Duration.ofSeconds(4), Duration.ofSeconds(30)), + batch -> { throw new AssertionError("missing vehicle facts must fail validation first"); }, + new ReadOnlyDispatchCycle(snapshot -> DispatchCycleResult.retrySnapshotChanged(), + new ShadowCommandIntentFactory()::create), gateway, + () -> ShadowEvidenceStorageHealth.Check.available(Long.MAX_VALUE), repository, + new ShadowCycleEvidenceCollector(UUID.randomUUID(), 1L, new ObjectMapper().findAndRegisterModules()), + report, clock); + scheduler = new Ft546ShadowScheduler(reader, cycle, report, clock, Duration.ofSeconds(1), state -> { + if (failPendingHeartbeat && Boolean.TRUE.equals(state.get("awaitingCommitCompletion"))) { + throw new IllegalStateException("heartbeat unavailable"); + } + heartbeats.add(state); + }); + } + + public void close() { repository.release.countDown(); scheduler.close(); } + } + + private static final class CommitRepository implements ShadowEvidenceRepository { + final CountDownLatch entered = new CountDownLatch(1); + final CountDownLatch release = new CountDownLatch(1); + final CountDownLatch finished = new CountDownLatch(1); + final JdbcTemplate jdbc; + final TransactionTemplate transaction; + + CommitRepository(CommitResult outcome) { + DriverManagerDataSource data = new DriverManagerDataSource( + "jdbc:h2:mem:" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1", "sa", ""); + jdbc = new JdbcTemplate(data); + jdbc.execute("CREATE TABLE committed_evidence (id INT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY)"); + AtomicBoolean first = new AtomicBoolean(true); + transaction = new TransactionTemplate(new DataSourceTransactionManager(data) { + @Override protected void doCommit(DefaultTransactionStatus status) { + if (!first.compareAndSet(true, false)) { super.doCommit(status); return; } + entered.countDown(); + boolean interrupted = false; + while (release.getCount() != 0) { + try { release.await(); } catch (InterruptedException ignored) { interrupted = true; } + } + try { + if (outcome == CommitResult.ROLLBACK) super.doRollback(status); + else super.doCommit(status); + if (outcome != CommitResult.SUCCESS) throw new TransactionSystemException("commit result unavailable"); + } finally { + if (interrupted) Thread.currentThread().interrupt(); + finished.countDown(); + } + } + }); + } + + int rows() { return jdbc.queryForObject("SELECT COUNT(*) FROM committed_evidence", Integer.class); } + public T inTransaction(Supplier work) { + return transaction.execute(status -> { jdbc.update("INSERT INTO committed_evidence DEFAULT VALUES"); return work.get(); }); + } + public SavedCycle save(ShadowCycleEvidence evidence) { return saveAll(Collections.singletonList(evidence)).get(0); } + public List saveAll(List evidence) { + List result = new ArrayList(); + for (ShadowCycleEvidence item : evidence) result.add(new SavedCycle(item.getCycleId(), item.getEvidenceHash(), item.getStatus(), item.getReasonCode())); + return result; + } + public SavedCycle findCycle(UUID runId, String hash, String task, String phase, long version) { return null; } + public ShadowEvidenceDigest dailyDigest(LocalDate date, ZoneId zone) { throw new UnsupportedOperationException(); } + public boolean isHealthy() { return true; } + } + + private static final class MutableClock extends Clock { + private volatile Instant now = NOW; + void advance(Duration duration) { now = now.plus(duration); } + public ZoneId getZone() { return ZoneOffset.UTC; } + public Clock withZone(ZoneId zone) { return this; } + public Instant instant() { return now; } + } +} diff --git a/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/config/ShadowEvidenceTimeoutTest.java b/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/config/ShadowEvidenceTimeoutTest.java new file mode 100644 index 0000000..326e273 --- /dev/null +++ b/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/config/ShadowEvidenceTimeoutTest.java @@ -0,0 +1,30 @@ +package com.ruoyi.shadow.config; + +import com.microsoft.sqlserver.jdbc.SQLServerDriver; +import com.zaxxer.hikari.HikariDataSource; +import org.junit.jupiter.api.Test; + +import java.sql.DriverPropertyInfo; +import java.util.HashMap; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.*; + +class ShadowEvidenceTimeoutTest { + @Test + void evidenceDriverHasFiniteIoAndCancellationLimitsEvenWhenUrlRequestsUnlimited() throws Exception { + ShadowProperties properties = new ShadowProperties(); + properties.getEvidence().setJdbcUrl("jdbc:sqlserver://localhost;databaseName=shadow;socketTimeout=0;cancelQueryTimeout=-1"); + try (HikariDataSource evidence = (HikariDataSource) new ShadowEvidenceConfiguration().shadowEvidenceDataSource(properties)) { + Map effective = new HashMap(); + for (DriverPropertyInfo info : new SQLServerDriver().getPropertyInfo( + evidence.getJdbcUrl(), evidence.getDataSourceProperties())) effective.put(info.name, info.value); + int socketMillis = Integer.parseInt(effective.get("socketTimeout")); + int cancelSeconds = Integer.parseInt(effective.get("cancelQueryTimeout")); + assertTrue(socketMillis > 10_000 && socketMillis <= 15_000, + "commit I/O must have a finite bound with room to report a late successful commit"); + assertTrue(cancelSeconds > 0 && cancelSeconds <= 2, + "query cancellation must not wait indefinitely after the statement deadline"); + } + } +} diff --git a/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/kernel/ShadowCycleDiagnosticsTest.java b/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/kernel/ShadowCycleDiagnosticsTest.java index f9e6eb9..463a869 100644 --- a/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/kernel/ShadowCycleDiagnosticsTest.java +++ b/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/kernel/ShadowCycleDiagnosticsTest.java @@ -27,6 +27,28 @@ class ShadowCycleDiagnosticsTest { assertEquals(6L, phases.get("TRANSACTION_COMMIT")); } + @Test + void observingPendingCommitAccumulatesTimeWithoutCancellingOrDoubleCounting() { + AtomicLong nanos = new AtomicLong(); + ShadowCycleDiagnostics diagnostics = new ShadowCycleDiagnostics(nanos::get); + nanos.set(2_000_000L); + diagnostics.enter("TRANSACTION_COMMIT"); + nanos.set(5_000_000L); + Map first = diagnostics.snapshot("AWAITING_COMMIT", null); + nanos.set(8_000_000L); + diagnostics.snapshot("AWAITING_COMMIT", null); + nanos.set(11_000_000L); + diagnostics.enter("RECORDING"); + nanos.set(13_000_000L); + Map completed = diagnostics.finish("COMMIT_CONFIRMED", null); + Map phases = (Map) completed.get("phaseMillis"); + assertEquals(13L, completed.get("elapsedMillis")); + assertEquals(2L, phases.get("QUEUED")); + assertEquals(9L, phases.get("TRANSACTION_COMMIT")); + assertEquals(2L, phases.get("RECORDING")); + assertEquals(3L, ((Map) first.get("phaseMillis")).get("TRANSACTION_COMMIT")); + } + @Test void timeoutSnapshotIsFrozenAndLateWorkerCannotEnterPersistence() { ShadowCycleDiagnostics diagnostics = new ShadowCycleDiagnostics();