diff --git a/docs/runbooks/ft546-shadow-20260910-source-recovery.md b/docs/runbooks/ft546-shadow-20260910-source-recovery.md new file mode 100644 index 0000000..d7cd038 --- /dev/null +++ b/docs/runbooks/ft546-shadow-20260910-source-recovery.md @@ -0,0 +1,51 @@ +# 2026-09-10 影子采集恢复修复 + +## 修复目标与现场事实 + +本次只修复影子模块的采集恢复及故障留证。新算法的分车、路径、等待时机可以与正式 WCS 不同，不以逐条一致或差异为零作为正常采集、算法可用或正式接管的门槛。 + +更新前影子源码为 `4408958`，buildId 为 `ft546-shadow-20260909-observed-traffic-v1`。现场 runId `27ff7d71-7074-4b6b-b5ac-de92c048e5ad` 的最后成功采集时间为北京时间 2026-09-10 10:10:11；状态为 `DEGRADED`、`collectionStopped=true`、`SOURCE_TRANSACTION_FAILED`，心跳仍在更新。该次总耗时约 7750 ms，没有超过 10000 ms 周期上限。旧版本未持久化异常子类型，不能确定本次究竟发生在获取连接、提交还是回滚阶段。 + +已通过故障注入确认的代码问题是：获取连接失败与明确已关闭连接的事务完成失败，原先都归为不可恢复的事务错误；另外，首次读取地图后提交失败会留下未成功事务的地图缓存。 + +## 行为变化 + +- 获取连接失败标为 `SOURCE_CONNECTION_UNAVAILABLE`，按 1、2、4、8、30 秒退避，成功后重置退避。 +- 只读事务提交或回滚失败时，只有当前连接明确 `isClosed()==true` 才标为 `SOURCE_CONNECTION_LOST`，完成本地清理后重读完整批次。失败批次不进入证据事务。 +- 连接仍打开、无法取得连接状态或检查状态本身失败时，继续按 `SOURCE_TRANSACTION_FAILED` 停采。不可取消的读取和证据存储故障的原有保护不变。 +- 地图缓存只在来源事务成功返回后发布，失败事务不会污染后续读取。 +- `/shadow/status` 和心跳增加 `lastCollectionFailure`，记录有界的异常类型、SQLState、错误码和首个栈位置；不记录原始异常消息、SQL 或连接字符串。恢复后保留上次失败诊断，当前状态仍正常更新。 +- 影子 profile 默认记录 `log/ft546-shadow.log`，10 MB 滚动、归档总量上限 50 MB、最多保留 5 天。现有完整异常日志可能包含异常消息，分享日志前仍需检查敏感内容。 + +新 buildId：`ft546-shadow-20260910-source-recovery-v1`。 + +不需要 SQL 迁移，不更改生产库的数据、表、视图或权限，不更改正式 WCS 调度及 PLC/WMS 通信。来源连接继续使用现场只读账户、原有限时和行数上限。 + +## 本地验证 + +在 `optimize/wcs` 工作树运行影子模块及依赖模块的离线测试，并加载已保存的现场输入： + +```text +mvn -o -pl ruoyi-wcs-shadow -am -Dshadow.replay.current-directory=D:/workSpace/PROJECT_SPACE/FT546/outputs/wcs-shadow-review-20260909 -Dshadow.replay.occupancy-directory=D:/workSpace/PROJECT_SPACE/FT546/outputs/wcs-shadow-review-20260909/post-update-inputs test +``` + +结果：`BUILD SUCCESS`，总计 355 项，353 项通过，2 项因未提供另一组旧归档输入而跳过，0 失败、0 错误。包括连接获取失败恢复、提交/回滚的关闭/打开/未知三态、重试重置、失败批次隔离、地图缓存发布时机，以及诊断留存及脱敏。 + +11 份当前现场输入和 50 份占用输入的回放通过；这证明已有回归断言成立，不等于任务已由影子算法完成。历史资源账本未重建，不能用这些回放中的 GRANTED/HOLD 数量直接推导现场完成率。 + +详细测试输出：工作区 `outputs/shadow-source-recovery-full-tests.log`。本地测试使用 Java 17；现场需使用 Java 8 编译生产源码。测试 H2 依赖需要较新 Java，不要求现场用 Java 8 执行该测试集。 + +## 现场交付与启动后验收 + +维护目标仅为 `C:\project\FT546-WCS-SHANDOW` 中的 `Ft546ShadowApplication`。更新前核实影子主类、端口和停止采集状态，保留现场 IDEA 配置及其他已有工作区修改。正式 WCS 不停机。现场采用 Java 8、Maven 3.6.0 和现有离线依赖库编译；完成后由操作人员在 IDEA 启动影子主类。 + +独立只读采集探针只验证当前连接、新采集器、完整批次及输入校验，不启动完整影子调度，不写入任一数据库。其成功不能替代重启后实际证据落库和持续运行检查。 + +启动后确认： + +1. buildId 为本次版本，13 项启动检查全部 PASS，`collectionStopped=false`。 +2. 间隔多个采集周期观察 `lastCompletedCycle` 持续推进、`evidenceBacklog` 不积压，新的完整输入及哈希、周期证据持续落库。 +3. 发生短时连接故障后能恢复，诊断能解释故障；无法确认事务安全结束的故障仍需排查。 +4. 分别评估输入完整性、路线和资源安全、可达性、等待解除和长期饥饿。新旧算法结果差异用于解释行为，不能直接当作算法错误；`SOURCE_GAP` 也可能是正式 WCS 尚未生成相应 `path_info`，需结合原始输入解释。 + +本次是恢复影子测试的数据采集，不是允许新算法接管正式调度。持续采集与有限快照回放不能单独证明新算法完成了现场任务。 diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/config/SqlServerReadOnlyTransactionManager.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/config/SqlServerReadOnlyTransactionManager.java index 42fd7c2..5d486cb 100644 --- a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/config/SqlServerReadOnlyTransactionManager.java +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/config/SqlServerReadOnlyTransactionManager.java @@ -1,10 +1,16 @@ package com.ruoyi.shadow.config; +import com.ruoyi.shadow.facts.ShadowSourceConnectionClosedException; +import org.springframework.jdbc.datasource.ConnectionHolder; import org.springframework.jdbc.datasource.DataSourceTransactionManager; import org.springframework.transaction.TransactionDefinition; +import org.springframework.transaction.TransactionSystemException; +import org.springframework.transaction.support.DefaultTransactionStatus; +import org.springframework.transaction.support.TransactionSynchronizationManager; import javax.sql.DataSource; import java.sql.Connection; +import java.sql.SQLException; /** * Rejects write-capable transaction definitions without emitting the @@ -35,4 +41,38 @@ final class SqlServerReadOnlyTransactionManager // Server does not support SET TRANSACTION READ ONLY. The source login // is independently verified to have no production write permission. } + + @Override + protected void doCommit(DefaultTransactionStatus status) { + try { + super.doCommit(status); + } catch (TransactionSystemException failure) { + throw classifyCompletionFailure(failure); + } + } + + @Override + protected void doRollback(DefaultTransactionStatus status) { + try { + super.doRollback(status); + } catch (TransactionSystemException failure) { + throw classifyCompletionFailure(failure); + } + } + + private TransactionSystemException classifyCompletionFailure(TransactionSystemException failure) { + Object resource = TransactionSynchronizationManager.getResource(obtainDataSource()); + if (resource instanceof ConnectionHolder) { + try { + // Do not turn an unknown/possibly live transaction into a retry. Spring + // still unbinds and releases this connection before the exception escapes. + if (((ConnectionHolder) resource).getConnection().isClosed()) { + return new ShadowSourceConnectionClosedException(failure); + } + } catch (SQLException | RuntimeException probeFailure) { + failure.addSuppressed(probeFailure); + } + } + return failure; + } } diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowSourceConnectionClosedException.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowSourceConnectionClosedException.java new file mode 100644 index 0000000..14e6eb4 --- /dev/null +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowSourceConnectionClosedException.java @@ -0,0 +1,10 @@ +package com.ruoyi.shadow.facts; + +import org.springframework.transaction.TransactionSystemException; + +/** A failed read-only transaction whose JDBC connection is confirmed closed. */ +public final class ShadowSourceConnectionClosedException extends TransactionSystemException { + public ShadowSourceConnectionClosedException(TransactionSystemException cause) { + super("SOURCE_CONNECTION_CLOSED", cause); + } +} 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 d3cdc0e..c3fd082 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 @@ -4,6 +4,7 @@ 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.CannotCreateTransactionException; import org.springframework.transaction.support.TransactionTemplate; import java.sql.PreparedStatement; @@ -75,6 +76,9 @@ public final class SqlServerShadowFactReader { throw new ShadowFactReadException("SOURCE_EMPTY_BATCH", Duration.ZERO); } + if (cachedMaps == null) { + cachedMaps = batch.getMaps(); + } consecutiveFailures.set(0); return batch; } catch (ShadowFactReadException failure) { @@ -85,6 +89,13 @@ public final class SqlServerShadowFactReader { throw failed(failure.getReasonCode(), failure); } catch (DataAccessException failure) { throw failed(classify(failure), failure); + } catch (CannotCreateTransactionException failure) { + // No collection callback ran; a temporary pool/connect failure can recover. + throw failed("SOURCE_CONNECTION_UNAVAILABLE", failure); + } catch (ShadowSourceConnectionClosedException failure) { + // The read-only manager confirmed closure and completed local cleanup. + // Discard all facts from this failed transaction and retry a fresh batch. + throw failed("SOURCE_CONNECTION_LOST", failure); } catch (TransactionException failure) { // A failed commit/rollback cannot be treated as a completed query timeout. throw failed("SOURCE_TRANSACTION_FAILED", failure); @@ -117,9 +128,6 @@ public final class SqlServerShadowFactReader { ensureTime(deadline); ShadowFactBatch batch = ShadowFactBatch.create(databaseStartedAt, databaseCompletedAt, queues, vehicles, maps); - if (cachedMaps == null) { - cachedMaps = maps; - } return batch; } 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 cd37bfd..575c6df 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 @@ -247,6 +247,7 @@ public final class Ft546ShadowScheduler implements AutoCloseable { Duration backoff = readFailure.getRetryAfter().isZero() ? CRASH_BACKOFF : readFailure.getRetryAfter(); statusReport.markDegraded(reason, failedAt); nextAttemptAt = failedAt.plus(backoff); + statusReport.recordCollectionFailure(reason, "SOURCE_READ", readFailure, failedAt, nextAttemptAt); LOGGER.warn("FT546 shadow source read failed: {}; retry after {} ms", reason, backoff.toMillis()); if (readFailure.getCause() != null) { @@ -264,6 +265,7 @@ public final class Ft546ShadowScheduler implements AutoCloseable { Instant failedAt = clock.instant(); statusReport.markDegraded("SHADOW_CYCLE_FAILED", failedAt); nextAttemptAt = failedAt.plus(CRASH_BACKOFF); + statusReport.recordCollectionFailure("SHADOW_CYCLE_FAILED", activeStage.name(), failure, failedAt, nextAttemptAt); LOGGER.error("FT546 shadow cycle failed", runtimeFailure); return false; } @@ -293,6 +295,7 @@ public final class Ft546ShadowScheduler implements AutoCloseable { permanentlyStopped.set(true); nextAttemptAt = Instant.EPOCH; statusReport.markPermanentlyDegraded(reasonCode, clock.instant()); + statusReport.recordCollectionFailure(reasonCode, stage.name(), failure, clock.instant(), null); LOGGER.error("FT546 shadow collection stopped: {}; stage={}; diagnostics={}", reasonCode, stage, statusReport.getCycleDiagnostics(), failure); return false; diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/startup/ShadowCollectionFailure.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/startup/ShadowCollectionFailure.java new file mode 100644 index 0000000..c0e4bfd --- /dev/null +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/startup/ShadowCollectionFailure.java @@ -0,0 +1,60 @@ +package com.ruoyi.shadow.startup; + +import org.springframework.transaction.TransactionSystemException; + +import java.sql.SQLException; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Collections; +import java.util.IdentityHashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** Bounded diagnostic facts; exception messages/SQL/connection strings are deliberately excluded. */ +final class ShadowCollectionFailure { + private ShadowCollectionFailure() { } + + static Map describe(String reason, String stage, Throwable failure, + Instant occurredAt, Instant nextAttemptAt) { + Map detail = new LinkedHashMap(); + detail.put("reasonCode", reason); + detail.put("stage", stage); + detail.put("occurredAt", occurredAt.toString()); + detail.put("retryable", nextAttemptAt != null); + detail.put("nextAttemptAt", nextAttemptAt == null ? null : nextAttemptAt.toString()); + List> exceptions = new ArrayList>(); + Set seen = Collections.newSetFromMap(new IdentityHashMap()); + collect(failure, "failure", exceptions, seen); + detail.put("exceptions", Collections.unmodifiableList(exceptions)); + return Collections.unmodifiableMap(detail); + } + + private static void collect(Throwable failure, String relation, List> exceptions, + Set seen) { + if (failure == null || exceptions.size() >= 12 || !seen.add(failure)) return; + Map entry = new LinkedHashMap(); + entry.put("type", failure.getClass().getName()); + entry.put("relation", relation); + if (failure instanceof SQLException) { + SQLException sql = (SQLException) failure; + entry.put("sqlState", sql.getSQLState()); + entry.put("sqlCode", sql.getErrorCode()); + } + StackTraceElement[] stack = failure.getStackTrace(); + if (stack.length > 0) entry.put("at", stack[0].toString()); + exceptions.add(Collections.unmodifiableMap(entry)); + collect(failure.getCause(), "cause", exceptions, seen); + if (failure instanceof TransactionSystemException) { + collect(((TransactionSystemException) failure).getApplicationException(), "application", exceptions, seen); + } + if (failure instanceof SQLException) { + collect(((SQLException) failure).getNextException(), "nextSql", exceptions, seen); + } + Throwable[] suppressed = failure.getSuppressed(); + for (int i = 0; i < Math.min(4, suppressed.length); i++) { + collect(suppressed[i], "suppressed", exceptions, seen); + } + } +} 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 6057bb1..04c5dbf 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-20260909-observed-traffic-v1"; + public static final String BUILD_ID = "ft546-shadow-20260910-source-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 7cdb8d0..bea125b 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 @@ -18,6 +18,7 @@ public final class ShadowStartupReport { private final Map startupChecks; private volatile Map cycleDiagnostics = Collections.emptyMap(); private volatile Map runtimeIdentity = Collections.emptyMap(); + private volatile Map lastCollectionFailure = Collections.emptyMap(); private ShadowStartupReport(Status status, Instant checkedAt, Instant lastCompletedCycle, @@ -48,6 +49,11 @@ public final class ShadowStartupReport { public Map getStartupChecks() { return startupChecks; } public Map getCycleDiagnostics() { return cycleDiagnostics; } public Map getRuntimeIdentity() { return runtimeIdentity; } + public Map getLastCollectionFailure() { return lastCollectionFailure; } + public void recordCollectionFailure(String reason, String stage, Throwable failure, + Instant occurredAt, Instant nextAttemptAt) { + lastCollectionFailure = ShadowCollectionFailure.describe(reason, stage, failure, occurredAt, nextAttemptAt); + } public void recordRuntimeIdentity(Map identity) { runtimeIdentity = Collections.unmodifiableMap(new LinkedHashMap(identity)); } @@ -113,6 +119,7 @@ public final class ShadowStartupReport { state.put("sourceLagMillis", sourceLagMillis); state.put("cycleDiagnostics", cycleDiagnostics); state.put("runtimeIdentity", runtimeIdentity); + state.put("lastCollectionFailure", lastCollectionFailure); return Collections.unmodifiableMap(state); } diff --git a/ruoyi-wcs-shadow/src/main/resources/application-shadow.yml b/ruoyi-wcs-shadow/src/main/resources/application-shadow.yml index c06690e..3647c47 100644 --- a/ruoyi-wcs-shadow/src/main/resources/application-shadow.yml +++ b/ruoyi-wcs-shadow/src/main/resources/application-shadow.yml @@ -4,6 +4,14 @@ server: spring: main: banner-mode: "off" +logging: + file: + name: log/ft546-shadow.log + logback: + rollingpolicy: + max-file-size: 10MB + total-size-cap: 50MB + max-history: 5 shadow: enabled: false traffic: 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 445cad1..5a482bf 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-20260909-observed-traffic-v1")) + .andExpect(jsonPath("$.runtimeIdentity.buildId").value("ft546-shadow-20260910-source-recovery-v1")) .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 606f27d..b4b0a08 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 @@ -25,6 +25,7 @@ import com.ruoyi.shadow.kernel.Ft546ShadowScheduler; import com.ruoyi.shadow.kernel.RecordingFt546CommandGateway; import com.ruoyi.shadow.kernel.ShadowCommandIntentFactory; import com.ruoyi.shadow.startup.ShadowStartupReport; +import com.ruoyi.shadow.config.ShadowSourceConfiguration; import com.ruoyi.wcs.application.ReadOnlyDispatchCycle; import com.ruoyi.wcs.application.dispatch.DispatchCycleResult; import org.junit.jupiter.api.Test; @@ -41,6 +42,10 @@ import org.springframework.transaction.support.TransactionTemplate; import org.slf4j.LoggerFactory; import java.time.Clock; +import java.sql.Connection; +import java.sql.SQLException; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Proxy; import java.time.Duration; import java.time.Instant; import java.time.LocalDate; @@ -640,6 +645,164 @@ class Ft546ShadowFaultIsolationTest { } } + @Test + void sourceFailureDetailsSurviveRecoveryWithoutExposingConnectionMessages() { + MutableClock clock = new MutableClock(NOW); + AtomicInteger reads = new AtomicInteger(); + SqlServerShadowFactReader.FactSource source = new FactSourceAdapter() { + @Override public Instant readDatabaseClock(Instant deadline) { + if (reads.incrementAndGet() == 1) { + throw new DataAccessResourceFailureException("private JDBC connection details", + new SQLException("private SQL connection details", "08S01", 10054)); + } + 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), clock), report, clock)) { + assertFalse(scheduler.runOnce()); + Map heartbeat = report.snapshotRuntimeState(clock.instant(), NOW, false); + assertTrue(heartbeat.get("lastCollectionFailure") instanceof Map, + "heartbeat must retain the failure even when IDEA does not save console output"); + Map detail = (Map) heartbeat.get("lastCollectionFailure"); + assertEquals("SOURCE_UNAVAILABLE", detail.get("reasonCode")); + assertEquals(Boolean.TRUE, detail.get("retryable")); + assertEquals(NOW.plusSeconds(1).toString(), detail.get("nextAttemptAt")); + String text = detail.toString(); + assertTrue(text.contains("08S01")); + assertTrue(text.contains("10054")); + assertTrue(text.contains("java.sql.SQLException")); + assertFalse(text.contains("private"), "raw exception messages must stay out of status data"); + clock.advance(Duration.ofSeconds(1)); + assertTrue(scheduler.runOnce()); + assertEquals(ShadowStartupReport.Status.READY, report.getStatus()); + assertEquals(detail, report.snapshotRuntimeState(clock.instant(), NOW, false).get("lastCollectionFailure")); + Map statusJson = new ObjectMapper().findAndRegisterModules().convertValue(report, Map.class); + assertEquals(detail, statusJson.get("lastCollectionFailure")); + } + } + + @Test + void connectionAcquisitionFailureRetriesWithoutCollectingPartialFacts() { + MutableClock clock = new MutableClock(NOW); + AtomicInteger connections = new AtomicInteger(); + AtomicInteger reads = new AtomicInteger(); + DriverManagerDataSource dataSource = new DriverManagerDataSource( + "jdbc:h2:mem:" + UUID.randomUUID(), "sa", "") { + @Override public Connection getConnection() throws SQLException { + int attempt = connections.incrementAndGet(); + if (attempt == 1 || attempt == 2 || attempt == 4) { + throw new java.sql.SQLTransientConnectionException("pool unavailable", "08001"); + } + return super.getConnection(); + } + }; + RecordingEvidenceRepository evidence = new RecordingEvidenceRepository(); + ShadowStartupReport report = report(); + SqlServerShadowFactReader source = new SqlServerShadowFactReader(emptySource(reads), + new TransactionTemplate(new ShadowSourceConfiguration().shadowSourceTx(dataSource)), clock); + try (Ft546ShadowScheduler scheduler = new Ft546ShadowScheduler(source, + evidenceCycle(report, evidence, gateway(), + () -> ShadowEvidenceStorageHealth.Check.available(Long.MAX_VALUE), clock), report, clock)) { + assertFalse(scheduler.runOnce()); + assertEquals("SOURCE_CONNECTION_UNAVAILABLE", report.getReasonCode()); + assertFalse(scheduler.isPermanentlyStopped()); + assertEquals(Duration.ofSeconds(1), scheduler.getRemainingBackoff()); + assertFalse(scheduler.runOnce()); + assertEquals(1, connections.get()); + assertEquals(0, reads.get()); + assertEquals(0, evidence.transactionCalls.get()); + clock.advance(Duration.ofSeconds(1)); + assertFalse(scheduler.runOnce()); + assertEquals(Duration.ofSeconds(2), scheduler.getRemainingBackoff()); + clock.advance(Duration.ofSeconds(2)); + assertTrue(scheduler.runOnce()); + assertEquals(ShadowStartupReport.Status.READY, report.getStatus()); + assertEquals(clock.instant(), report.getLastCompletedCycle()); + assertEquals(1, evidence.transactionCalls.get()); + assertFalse(scheduler.runOnce()); + assertEquals(Duration.ofSeconds(1), scheduler.getRemainingBackoff()); + clock.advance(Duration.ofSeconds(1)); + assertTrue(scheduler.runOnce()); + assertEquals(2, evidence.transactionCalls.get()); + } + } + + @Test + void sourceCompletionFailureRetriesOnlyWhenConnectionIsConfirmedClosed() { + for (boolean rollback : new boolean[] {false, true}) { + for (int connectionState : new int[] {0, 1, 2}) { + boolean closed = connectionState == 1; + boolean unknown = connectionState == 2; + MutableClock clock = new MutableClock(NOW); + AtomicInteger connections = new AtomicInteger(); + AtomicReference failedConnection = new AtomicReference(); + DriverManagerDataSource dataSource = new DriverManagerDataSource( + "jdbc:h2:mem:" + UUID.randomUUID(), "sa", "") { + @Override public Connection getConnection() throws SQLException { + Connection delegate = super.getConnection(); + if (connections.incrementAndGet() != 1) return delegate; + failedConnection.set(delegate); + return (Connection) Proxy.newProxyInstance(Connection.class.getClassLoader(), + new Class[] {Connection.class}, (proxy, method, args) -> { + if (unknown && "isClosed".equals(method.getName())) { + throw new SQLException("connection state unavailable"); + } + if ((rollback ? "rollback" : "commit").equals(method.getName())) { + if (closed) delegate.close(); + throw new SQLException("transaction completion failed", "08006"); + } + try { return method.invoke(delegate, args); } + catch (InvocationTargetException error) { throw error.getCause(); } + }); + } + }; + AtomicInteger reads = new AtomicInteger(); + AtomicInteger mapReads = new AtomicInteger(); + SqlServerShadowFactReader.FactSource facts = new FactSourceAdapter() { + @Override public Instant readDatabaseClock(Instant deadline) { + if (reads.incrementAndGet() == 1 && rollback) { + throw new QueryTimeoutException("query completed with timeout"); + } + return clock.instant(); + } + @Override public List readMaps(Instant deadline) { + mapReads.incrementAndGet(); + return Collections.emptyList(); + } + }; + RecordingEvidenceRepository evidence = new RecordingEvidenceRepository(); + ShadowStartupReport report = report(); + SqlServerShadowFactReader source = new SqlServerShadowFactReader(facts, + new TransactionTemplate(new ShadowSourceConfiguration().shadowSourceTx(dataSource)), clock); + try (Ft546ShadowScheduler scheduler = new Ft546ShadowScheduler(source, + evidenceCycle(report, evidence, gateway(), + () -> ShadowEvidenceStorageHealth.Check.available(Long.MAX_VALUE), clock), report, clock)) { + assertFalse(scheduler.runOnce()); + assertEquals(0, evidence.transactionCalls.get()); + assertEquals(!closed, scheduler.isPermanentlyStopped()); + assertEquals(closed ? "SOURCE_CONNECTION_LOST" : "SOURCE_TRANSACTION_FAILED", + report.getReasonCode()); + assertFalse(scheduler.runOnce()); + clock.advance(Duration.ofSeconds(1)); + assertEquals(closed, scheduler.runOnce()); + assertEquals(closed ? 1 : 0, evidence.transactionCalls.get()); + assertEquals(closed ? 2 : 1, connections.get()); + if (closed) { + assertEquals(ShadowStartupReport.Status.READY, report.getStatus()); + assertEquals(clock.instant(), report.getLastCompletedCycle()); + assertEquals(rollback ? 1 : 2, mapReads.get(), + "a failed source transaction must not publish its map cache"); + } + } + try { assertTrue(failedConnection.get().isClosed()); } + catch (SQLException impossible) { throw new AssertionError(impossible); } + } + } + } + @Test void failedSourceRollbackPermanentlyStopsEvenWhenOriginalCauseWasTimeout() { MutableClock clock = new MutableClock(NOW); @@ -737,6 +900,13 @@ class Ft546ShadowFaultIsolationTest { ShadowStartupReport report, RecordingEvidenceRepository evidence, RecordingFt546CommandGateway gateway, ShadowEvidenceStorageHealth storageHealth) { + return evidenceCycle(report, evidence, gateway, storageHealth, Clock.systemUTC()); + } + + private static Ft546ShadowCycle evidenceCycle( + ShadowStartupReport report, RecordingEvidenceRepository evidence, + RecordingFt546CommandGateway gateway, + ShadowEvidenceStorageHealth storageHealth, Clock clock) { return new Ft546ShadowCycle( new ShadowFactBatchValidator(Duration.ofSeconds(4), Duration.ofSeconds(30)), @@ -746,7 +916,7 @@ class Ft546ShadowFaultIsolationTest { new ShadowCommandIntentFactory()::create), gateway, storageHealth, evidence, new ShadowCycleEvidenceCollector(evidence.runId, 1L, - new ObjectMapper().findAndRegisterModules()), report); + new ObjectMapper().findAndRegisterModules()), report, clock); } private static RecordingFt546CommandGateway gateway() {