lhq/FT546-WCS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(shadow): recover closed source connections and retain diagnostics

Browse Source
This commit is contained in:
zgw
2026-09-10 11:55:59 +08:00
parent 4408958443
commit 0d53f41f1a
11 changed files with 363 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/Ft546ShadowApplicationTest.java
View File

@@ -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")

172
ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/Ft546ShadowFaultIsolationTest.java
View File

@@ -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<String, Object> 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<Connection> failedConnection = new AtomicReference<Connection>();
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<ShadowMapFact> 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() {