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

fix(shadow): harden production observation cycle

Browse Source
This commit is contained in:
zgw
2026-09-03 11:10:22 +08:00
parent 0080befff4
commit 9829b9c0fa
7 changed files with 325 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/config/ShadowKernelConfiguration.java
View File

@@ -29,6 +29,7 @@ import com.ruoyi.wcs.domain.traffic.WaitForGraph;
import com.ruoyi.wcs.port.out.DispatchPolicy;
import com.ruoyi.wcs.port.out.TrafficStateRepository;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.JdbcTemplate;
@@ -184,8 +185,11 @@ public class ShadowKernelConfiguration {
SqlServerShadowFactReader factReader,
Ft546ShadowCycle shadowCycle,
ShadowStartupReport shadowStartupReport,
@Qualifier("shadowClock") Clock clock) {
@Qualifier("shadowClock") Clock clock,
@Value("${shadow.cycle.deadline-ms:10000}")
long cycleDeadlineMs) {
return new Ft546ShadowScheduler(factReader, shadowCycle,
shadowStartupReport, clock);
shadowStartupReport, clock,
Duration.ofMillis(cycleDeadlineMs));
}
}

15
ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowFactBatchValidator.java
View File

@@ -8,6 +8,9 @@ import java.util.Map;
import java.util.Set;
public final class ShadowFactBatchValidator {
private static final Duration MAXIMUM_FUTURE_SKEW =
Duration.ofMillis(100);
private final Duration maximumCaptureSpan;
private final Duration maximumAge;
@@ -38,7 +41,9 @@ public final class ShadowFactBatchValidator {
return ShadowFactValidation.insufficient(
"DATABASE_CAPTURE_SPAN_EXCEEDED");
}
if (completed.isAfter(now)) {
if (completed.isAfter(now)
&& Duration.between(now, completed)
.compareTo(MAXIMUM_FUTURE_SKEW) > 0) {
return ShadowFactValidation.insufficient("DATABASE_CLOCK_IN_FUTURE");
}
if (Duration.between(completed, now).compareTo(maximumAge) > 0) {
@@ -53,6 +58,7 @@ public final class ShadowFactBatchValidator {
Map<String, String> vehicleHashes = new HashMap<String, String>();
Map<String, ShadowVehicleFact> vehicles =
new HashMap<String, ShadowVehicleFact>();
// device_srgv.last_time is the last task completion time, not a heartbeat.
for (ShadowVehicleFact vehicle : batch.getVehicles()) {
if (!notBlank(vehicle.getCurrentSite())) {
return ShadowFactValidation.insufficient(
@@ -76,13 +82,6 @@ public final class ShadowFactBatchValidator {
return ShadowFactValidation.insufficient(
"CONTRADICTORY_VEHICLE_POSITION:" + vehicle.getDeviceId());
}
if (vehicle.getCapturedAt() == null
|| vehicle.getCapturedAt().isAfter(now)
|| Duration.between(vehicle.getCapturedAt(), now)
.compareTo(maximumAge) > 0) {
return ShadowFactValidation.insufficient(
"STALE_VEHICLE_FACT:" + vehicle.getDeviceId());
}
String existingHash = vehicleHashes.putIfAbsent(
vehicle.getDeviceId(), vehicle.getContentHash());
if (existingHash != null

2
ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/SqlServerShadowFactReader.java
View File

@@ -154,7 +154,7 @@ public final class SqlServerShadowFactReader {
private static String classify(DataAccessException failure) {
String name = failure.getClass().getSimpleName().toUpperCase();
if (name.contains("TIMEOUT") || name.contains("TRANSIENT")) {
if (name.contains("TIMEOUT")) {
return "SOURCE_QUERY_TIMEOUT";
}
return "SOURCE_UNAVAILABLE";

75
ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/kernel/Ft546ShadowScheduler.java
View File

@@ -26,7 +26,8 @@ import java.util.concurrent.atomic.AtomicBoolean;
public final class Ft546ShadowScheduler implements AutoCloseable {
private static final Logger LOGGER =
LoggerFactory.getLogger(Ft546ShadowScheduler.class);
private static final Duration CYCLE_DEADLINE = Duration.ofSeconds(4);
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 String FINISH_QUEUE_VIEW =
"shadow_ro.v_wcs_srgv_queue_finish";
@@ -36,9 +37,11 @@ public final class Ft546ShadowScheduler implements AutoCloseable {
private final ShadowStartupReport statusReport;
private final Clock clock;
private final Duration cycleDeadline;
private final Duration sourceReadDeadline;
private final ExecutorService cycleExecutor;
private final AtomicBoolean inFlight = new AtomicBoolean();
private final AtomicBoolean permanentlyStopped = new AtomicBoolean();
private volatile CycleStage activeStage = CycleStage.IDLE;
private volatile ShadowReadCursor cursor = ShadowReadCursor.initial();
private volatile Instant nextAttemptAt = Instant.EPOCH;
@@ -54,12 +57,23 @@ public final class Ft546ShadowScheduler implements AutoCloseable {
ShadowStartupReport statusReport,
Clock clock,
Duration cycleDeadline) {
this(factReader, shadowCycle, statusReport, clock, cycleDeadline,
shorter(cycleDeadline, SOURCE_READ_DEADLINE));
}
Ft546ShadowScheduler(SqlServerShadowFactReader factReader,
Ft546ShadowCycle shadowCycle,
ShadowStartupReport statusReport,
Clock clock,
Duration cycleDeadline,
Duration sourceReadDeadline) {
this.factReader = Objects.requireNonNull(factReader, "factReader");
this.shadowCycle = Objects.requireNonNull(shadowCycle, "shadowCycle");
this.statusReport = Objects.requireNonNull(statusReport,
"statusReport");
this.clock = Objects.requireNonNull(clock, "clock");
this.cycleDeadline = requirePositive(cycleDeadline);
this.sourceReadDeadline = requirePositive(sourceReadDeadline);
this.cycleExecutor = Executors.newSingleThreadExecutor(
shadowCycleThreadFactory());
}
@@ -94,16 +108,21 @@ public final class Ft546ShadowScheduler implements AutoCloseable {
nextAttemptAt = Instant.EPOCH;
return true;
} catch (TimeoutException timeout) {
CycleStage timedOutStage = activeStage;
future.cancel(true);
return stopPermanently("SHADOW_CYCLE_TIMEOUT", timeout);
return stopPermanently("SHADOW_CYCLE_TIMEOUT", timeout,
timedOutStage);
} catch (InterruptedException interrupted) {
CycleStage interruptedStage = activeStage;
future.cancel(true);
Thread.currentThread().interrupt();
return stopPermanently("SHADOW_CYCLE_INTERRUPTED", interrupted);
return stopPermanently("SHADOW_CYCLE_INTERRUPTED", interrupted,
interruptedStage);
} catch (ExecutionException wrapped) {
return handleFailure(wrapped.getCause());
} finally {
inFlight.set(false);
activeStage = CycleStage.IDLE;
}
}
@@ -123,15 +142,23 @@ public final class Ft546ShadowScheduler implements AutoCloseable {
}
private CycleAttempt executeCycle(Instant startedAt) {
activeStage = CycleStage.SOURCE_READ;
ShadowFactBatch batch = factReader.read(cursor,
startedAt.plus(cycleDeadline));
return new CycleAttempt(batch, shadowCycle.run(batch, clock.instant()));
startedAt.plus(sourceReadDeadline));
activeStage = CycleStage.SHADOW_CYCLE;
return new CycleAttempt(batch,
shadowCycle.run(batch, clock.instant()));
}
private boolean handleFailure(Throwable failure) {
if (failure instanceof ShadowFactReadException) {
ShadowFactReadException readFailure =
(ShadowFactReadException) failure;
String timeoutReason = sourceTimeoutReason(readFailure);
if (timeoutReason != null) {
return stopPermanently(timeoutReason, readFailure,
CycleStage.SOURCE_READ);
}
Instant failedAt = clock.instant();
statusReport.markDegraded(readFailure.getReasonCode(), failedAt);
nextAttemptAt = failedAt.plus(readFailure.getRetryAfter());
@@ -157,15 +184,41 @@ public final class Ft546ShadowScheduler implements AutoCloseable {
return false;
}
private boolean stopPermanently(String reasonCode, Exception failure) {
private static String sourceTimeoutReason(
ShadowFactReadException failure) {
String reasonCode = failure.getReasonCode();
if ("SOURCE_QUERY_TIMEOUT".equals(reasonCode)
|| "SOURCE_BATCH_DEADLINE_EXCEEDED".equals(reasonCode)) {
return reasonCode;
}
Throwable cause = failure.getCause();
while (cause != null) {
String type = cause.getClass().getSimpleName()
.toUpperCase(java.util.Locale.ROOT);
if (type.contains("TIMEOUT")) {
return "SOURCE_QUERY_TIMEOUT";
}
cause = cause.getCause();
}
return null;
}
private boolean stopPermanently(String reasonCode, Exception failure,
CycleStage stage) {
permanentlyStopped.set(true);
nextAttemptAt = Instant.EPOCH;
statusReport.markPermanentlyDegraded(reasonCode, clock.instant());
LOGGER.error("FT546 shadow collection stopped: {}", reasonCode,
failure);
LOGGER.error("FT546 shadow collection stopped: {}; stage={}",
reasonCode, stage, failure);
return false;
}
private static Duration shorter(Duration left, Duration right) {
Duration first = requirePositive(left);
Duration second = requirePositive(right);
return first.compareTo(second) <= 0 ? first : second;
}
private static Duration requirePositive(Duration value) {
Duration duration = Objects.requireNonNull(value, "cycleDeadline");
if (duration.isZero() || duration.isNegative()
@@ -184,6 +237,12 @@ public final class Ft546ShadowScheduler implements AutoCloseable {
};
}
private enum CycleStage {
IDLE,
SOURCE_READ,
SHADOW_CYCLE
}
private static ShadowReadCursor advanceFinishCursor(
ShadowReadCursor current, ShadowFactBatch batch) {
Instant latestDate = current.getLastCreateDate();

2
ruoyi-wcs-shadow/src/main/resources/application-shadow.yml
View File

@@ -6,6 +6,8 @@ spring:
banner-mode: "off"
shadow:
enabled: false
cycle:
deadline-ms: 10000
poll:
initial-delay-ms: 5000
fixed-delay-ms: 5000

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

@@ -1,5 +1,8 @@
package com.ruoyi.shadow;
import ch.qos.logback.classic.Logger;
import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.read.ListAppender;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.ruoyi.ft546.protocol.Ft546DeviceProfile;
import com.ruoyi.ft546.protocol.Ft546SrgvCommandEncoder;
@@ -11,6 +14,7 @@ import com.ruoyi.shadow.evidence.ShadowEvidenceException;
import com.ruoyi.shadow.evidence.ShadowEvidenceRepository;
import com.ruoyi.shadow.evidence.ShadowEvidenceStorageHealth;
import com.ruoyi.shadow.facts.ShadowFactBatchValidator;
import com.ruoyi.shadow.facts.ShadowFactReadException;
import com.ruoyi.shadow.facts.ShadowMapFact;
import com.ruoyi.shadow.facts.ShadowQueueFact;
import com.ruoyi.shadow.facts.ShadowReadCursor;
@@ -28,10 +32,13 @@ import org.springframework.core.io.ClassPathResource;
import org.springframework.core.env.EnumerablePropertySource;
import org.springframework.core.env.PropertySource;
import org.springframework.boot.env.YamlPropertySourceLoader;
import org.springframework.dao.DataAccessResourceFailureException;
import org.springframework.dao.QueryTimeoutException;
import org.springframework.dao.TransientDataAccessResourceException;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.jdbc.datasource.DriverManagerDataSource;
import org.springframework.transaction.support.TransactionTemplate;
import org.slf4j.LoggerFactory;
import java.time.Clock;
import java.time.Duration;
@@ -51,6 +58,7 @@ import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -61,7 +69,7 @@ class Ft546ShadowFaultIsolationTest {
private static final Instant NOW = Instant.parse("2026-08-14T06:00:00Z");
@Test
void sourceTimeoutNeverEntersCycleOrRecordsACommand() {
void sourceQueryTimeoutPermanentlyStopsCollectionBeforeCycle() {
MutableClock clock = new MutableClock(NOW);
AtomicInteger sourceCalls = new AtomicInteger();
SqlServerShadowFactReader reader = reader(timeoutSource(sourceCalls), clock);
@@ -72,15 +80,172 @@ class Ft546ShadowFaultIsolationTest {
evidenceCycle(report, evidence, gateway,
() -> ShadowEvidenceStorageHealth.Check.available(
Long.MAX_VALUE)), report, clock);
Logger schedulerLogger = (Logger) LoggerFactory.getLogger(
Ft546ShadowScheduler.class);
ListAppender<ILoggingEvent> logAppender =
new ListAppender<ILoggingEvent>();
logAppender.start();
schedulerLogger.addAppender(logAppender);
try {
assertFalse(scheduler.runOnce());
clock.advance(Duration.ofMinutes(1));
assertFalse(scheduler.runOnce());
assertEquals(1, sourceCalls.get());
assertEquals(0, evidence.transactionCalls.get());
assertEquals(Collections.emptyList(), gateway.getRecorded());
assertEquals(ShadowStartupReport.Status.DEGRADED, report.getStatus());
assertEquals(ShadowStartupReport.Status.DEGRADED,
report.getStatus());
assertEquals("SOURCE_QUERY_TIMEOUT", report.getReasonCode());
assertEquals(Duration.ofSeconds(1), scheduler.getRemainingBackoff());
assertTrue(scheduler.isPermanentlyStopped());
assertTrue(logAppender.list.stream().anyMatch(event ->
event.getFormattedMessage()
.contains("stage=SOURCE_READ")),
"source timeout log must identify the blocked stage");
} finally {
schedulerLogger.detachAppender(logAppender);
logAppender.stop();
scheduler.close();
}
}
@Test
void sourceBatchDeadlinePermanentlyStopsCollection() {
MutableClock clock = new MutableClock(NOW);
AtomicInteger sourceCalls = new AtomicInteger();
SqlServerShadowFactReader.FactSource source = new FactSourceAdapter() {
@Override
public Instant readDatabaseClock(Instant deadline) {
sourceCalls.incrementAndGet();
throw new ShadowFactReadException(
"SOURCE_BATCH_DEADLINE_EXCEEDED", Duration.ZERO);
}
};
ShadowStartupReport report = report();
Ft546ShadowScheduler scheduler = new Ft546ShadowScheduler(
reader(source, clock),
evidenceCycle(report, new RecordingEvidenceRepository(),
gateway(),
() -> ShadowEvidenceStorageHealth.Check.available(
Long.MAX_VALUE)),
report, clock);
try {
assertFalse(scheduler.runOnce());
clock.advance(Duration.ofMinutes(1));
assertFalse(scheduler.runOnce());
assertEquals(1, sourceCalls.get());
assertEquals("SOURCE_BATCH_DEADLINE_EXCEEDED",
report.getReasonCode());
assertTrue(scheduler.isPermanentlyStopped());
} finally {
scheduler.close();
}
}
@Test
void wrappedSourceQueryTimeoutPermanentlyStopsCollection() {
MutableClock clock = new MutableClock(NOW);
AtomicInteger sourceCalls = new AtomicInteger();
SqlServerShadowFactReader.FactSource source = new FactSourceAdapter() {
@Override
public Instant readDatabaseClock(Instant deadline) {
sourceCalls.incrementAndGet();
throw new DataAccessResourceFailureException(
"wrapped source timeout",
new QueryTimeoutException("query timed out"));
}
};
ShadowStartupReport report = report();
Ft546ShadowScheduler scheduler = new Ft546ShadowScheduler(
reader(source, clock),
evidenceCycle(report, new RecordingEvidenceRepository(),
gateway(),
() -> ShadowEvidenceStorageHealth.Check.available(
Long.MAX_VALUE)),
report, clock);
try {
assertFalse(scheduler.runOnce());
clock.advance(Duration.ofMinutes(1));
assertFalse(scheduler.runOnce());
assertEquals(1, sourceCalls.get());
assertEquals("SOURCE_QUERY_TIMEOUT", report.getReasonCode());
assertTrue(scheduler.isPermanentlyStopped());
} finally {
scheduler.close();
}
}
@Test
void nonTimeoutSourceFailureStillUsesBackoff() {
MutableClock clock = new MutableClock(NOW);
AtomicInteger sourceCalls = new AtomicInteger();
SqlServerShadowFactReader.FactSource source = new FactSourceAdapter() {
@Override
public Instant readDatabaseClock(Instant deadline) {
sourceCalls.incrementAndGet();
throw new DataAccessResourceFailureException(
"simulated source outage");
}
};
ShadowStartupReport report = report();
Ft546ShadowScheduler scheduler = new Ft546ShadowScheduler(
reader(source, clock),
evidenceCycle(report, new RecordingEvidenceRepository(),
gateway(),
() -> ShadowEvidenceStorageHealth.Check.available(
Long.MAX_VALUE)),
report, clock);
try {
assertFalse(scheduler.runOnce());
assertEquals(1, sourceCalls.get());
assertEquals("SOURCE_UNAVAILABLE", report.getReasonCode());
assertFalse(scheduler.isPermanentlyStopped());
assertEquals(Duration.ofSeconds(1),
scheduler.getRemainingBackoff());
} finally {
scheduler.close();
}
}
@Test
void transientSourceFailureStillUsesBackoff() {
MutableClock clock = new MutableClock(NOW);
AtomicInteger sourceCalls = new AtomicInteger();
SqlServerShadowFactReader.FactSource source = new FactSourceAdapter() {
@Override
public Instant readDatabaseClock(Instant deadline) {
sourceCalls.incrementAndGet();
throw new TransientDataAccessResourceException(
"simulated transient source failure");
}
};
ShadowStartupReport report = report();
Ft546ShadowScheduler scheduler = new Ft546ShadowScheduler(
reader(source, clock),
evidenceCycle(report, new RecordingEvidenceRepository(),
gateway(),
() -> ShadowEvidenceStorageHealth.Check.available(
Long.MAX_VALUE)),
report, clock);
try {
assertFalse(scheduler.runOnce());
assertEquals(1, sourceCalls.get());
assertEquals("SOURCE_UNAVAILABLE", report.getReasonCode());
assertFalse(scheduler.isPermanentlyStopped());
assertEquals(Duration.ofSeconds(1),
scheduler.getRemainingBackoff());
} finally {
scheduler.close();
}
}
@Test
@@ -174,6 +339,12 @@ class Ft546ShadowFaultIsolationTest {
}, "shadow-cycle-test-release");
delayedRelease.setDaemon(true);
delayedRelease.start();
Logger schedulerLogger = (Logger) LoggerFactory.getLogger(
Ft546ShadowScheduler.class);
ListAppender<ILoggingEvent> logAppender =
new ListAppender<ILoggingEvent>();
logAppender.start();
schedulerLogger.addAppender(logAppender);
long startedAt = System.nanoTime();
try {
@@ -189,6 +360,10 @@ class Ft546ShadowFaultIsolationTest {
assertEquals(ShadowStartupReport.Status.DEGRADED,
report.getStatus());
assertEquals("SHADOW_CYCLE_TIMEOUT", report.getReasonCode());
assertTrue(logAppender.list.stream().anyMatch(event ->
event.getFormattedMessage()
.contains("stage=SHADOW_CYCLE")),
"timeout log must identify the blocked stage");
assertFalse(scheduler.runOnce());
assertEquals(callsAfterTimeout, sourceCalls.get());
assertEquals(Collections.emptyList(), gateway.getRecorded());
@@ -198,6 +373,8 @@ class Ft546ShadowFaultIsolationTest {
report.getStatus());
assertEquals("SHADOW_CYCLE_TIMEOUT", report.getReasonCode());
} finally {
schedulerLogger.detachAppender(logAppender);
logAppender.stop();
releaseCycle.countDown();
scheduler.close();
}
@@ -219,10 +396,40 @@ class Ft546ShadowFaultIsolationTest {
assertTrue(configuredKeys.contains("shadow.source.jdbc-url"));
assertTrue(configuredKeys.contains("shadow.evidence.jdbc-url"));
assertTrue(configuredKeys.contains("shadow.cycle.deadline-ms"));
assertFalse(configuredKeys.stream().anyMatch(
Ft546ShadowFaultIsolationTest::isOutboundEndpointKey));
}
@Test
void longCycleDeadlineKeepsSourceReadDeadlineCappedAtFourSeconds() {
MutableClock clock = new MutableClock(NOW);
AtomicReference<Instant> observedDeadline =
new AtomicReference<Instant>();
SqlServerShadowFactReader.FactSource source = new FactSourceAdapter() {
@Override
public Instant readDatabaseClock(Instant deadline) {
observedDeadline.set(deadline);
return NOW;
}
};
RecordingEvidenceRepository evidence = new RecordingEvidenceRepository();
ShadowStartupReport report = report();
Ft546ShadowScheduler scheduler = new Ft546ShadowScheduler(
reader(source, clock),
evidenceCycle(report, evidence, gateway(),
() -> ShadowEvidenceStorageHealth.Check.available(
Long.MAX_VALUE)),
report, clock, Duration.ofSeconds(10));
try {
assertTrue(scheduler.runOnce());
assertEquals(NOW.plusSeconds(4), observedDeadline.get());
} finally {
scheduler.close();
}
}
@Test
void activeQueueCannotAdvanceTheFinishHistoryCursor() {
MutableClock clock = new MutableClock(NOW);

27
ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/facts/ShadowFactBatchValidatorTest.java
View File

@@ -28,6 +28,33 @@ class ShadowFactBatchValidatorTest {
assertEquals("FACTS_COMPLETE", result.getReasonCode());
}
@Test
void databaseClockOneMillisecondAheadIsComparable() {
ShadowFactValidation result = validator.validate(batch(
Collections.singletonList(queue("Q-1", "070501", "070502", "1")),
Collections.singletonList(vehicle("1", "070501")),
Arrays.asList(map("070501"), map("070502")),
NOW.minusSeconds(1), NOW.plusMillis(1)), NOW);
assertEquals(ShadowFactValidation.Status.COMPARABLE,
result.getStatus());
}
@Test
void oldLastTaskTimeDoesNotMakeVehicleFactStale() {
ShadowVehicleFact idleVehicle = completeVehicleBuilder("1", "070501")
.capturedAt(NOW.minus(Duration.ofHours(2)))
.build();
ShadowFactValidation result = validator.validate(batch(
Collections.singletonList(queue("Q-1", "070501", "070502", "1")),
Collections.singletonList(idleVehicle),
Arrays.asList(map("070501"), map("070502")),
NOW.minusSeconds(1), NOW.minusMillis(500)), NOW);
assertEquals(ShadowFactValidation.Status.COMPARABLE,
result.getStatus());
}
@Test
void contradictoryVehiclePositionsAreInsufficient() {
ShadowFactValidation result = validator.validate(batch(