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

fix(shadow): recover completed read timeouts and pair issued observations

Browse Source
This commit is contained in:
zgw
2026-09-09 10:00:43 +08:00
parent af8b5f1830
commit edb13450a6
14 changed files with 948 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/kernel/Ft546ShadowCycle.java
View File

@@ -18,6 +18,7 @@ import com.ruoyi.wcs.application.dispatch.DispatchSnapshot;
import java.time.Instant;
import java.time.Clock;
import java.util.List;
import java.util.ArrayList;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
@@ -110,14 +111,16 @@ public final class Ft546ShadowCycle {
return degrade(health.getReasonCode(), clock.instant());
}
try {
List<Runnable> afterCommit = new ArrayList<Runnable>();
progress.enter("TRANSACTION_BEGIN");
ShadowCycleResult result = evidenceRepository.inTransaction(
() -> {
ShadowCycleResult saved = runAndPersist(batch, now, progress);
ShadowCycleResult saved = runAndPersist(batch, now, progress, afterCommit);
progress.enter("TRANSACTION_COMMIT");
return saved;
});
progress.enter("RECORDING");
afterCommit.forEach(Runnable::run);
commandGateway.rememberAll(result.getEncodedCommands());
Instant completedAt = clock.instant();
Long lag = batch.getDatabaseCapturedAt() == null ? null
@@ -127,12 +130,13 @@ public final class Ft546ShadowCycle {
statusReport.markReady(completedAt, lag);
return result;
} catch (RuntimeException evidenceFailure) {
invalidatePendingComparisons();
return degrade(failureReason(evidenceFailure), clock.instant());
}
}
private ShadowCycleResult runAndPersist(ShadowFactBatch batch,
Instant now, ShadowCycleDiagnostics progress) {
Instant now, ShadowCycleDiagnostics progress, List<Runnable> afterCommit) {
progress.enter("VALIDATION");
ShadowFactValidation validation = validator.validate(batch, now);
if (validation.getStatus()
@@ -169,16 +173,18 @@ public final class Ft546ShadowCycle {
}
progress.enter("EVIDENCE_BUILD");
progress.dispatch(batch, snapshot, result.getDecisions());
List<ShadowCycleEvidence> evidence = evidenceCollector.collect(batch,
ShadowCycleEvidenceCollector.CollectedEvidence evidence = evidenceCollector.prepare(batch,
result.getDecisions(), encoded, clock.instant());
progress.enter("EVIDENCE_SAVE");
evidenceRepository.saveAll(evidence);
List<ShadowEvidenceRepository.SavedCycle> saved = evidenceRepository.saveAll(evidence.getEvidence());
afterCommit.add(() -> evidence.committed(saved));
return ShadowCycleResult.comparable(result.getDecisions(),
result.getIntents(), encoded);
}
private ShadowCycleResult persistInsufficient(
ShadowFactBatch batch, String reasonCode, ShadowCycleDiagnostics progress) {
invalidatePendingComparisons();
progress.enter("EVIDENCE_BUILD");
List<ShadowCycleEvidence> evidence = evidenceCollector.insufficient(batch,
reasonCode, clock.instant());
@@ -188,6 +194,7 @@ public final class Ft546ShadowCycle {
}
private ShadowCycleResult degrade(String reasonCode, Instant now) {
invalidatePendingComparisons();
degradedReason.compareAndSet(null, reasonCode);
String stoppedReason = degradedReason.get();
statusReport.markDegraded(stoppedReason, now);
@@ -209,6 +216,10 @@ public final class Ft546ShadowCycle {
return "EVIDENCE_STORAGE_UNAVAILABLE";
}
public void invalidatePendingComparisons() {
if (evidenceCollector != null) evidenceCollector.invalidatePendingComparisons();
}
private ShadowCycleResult runWithoutEvidence(ShadowFactBatch batch,
Instant now, ShadowCycleDiagnostics progress) {
progress.enter("VALIDATION");

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

@@ -114,6 +114,7 @@ public final class Ft546ShadowScheduler implements AutoCloseable {
}
public boolean runOnce() {
long startedNanos = System.nanoTime();
Instant now = clock.instant();
if (permanentlyStopped.get() || now.isBefore(nextAttemptAt)
|| !inFlight.compareAndSet(false, true)) {
@@ -121,6 +122,7 @@ public final class Ft546ShadowScheduler implements AutoCloseable {
}
ShadowCycleDiagnostics diagnostics = new ShadowCycleDiagnostics();
lastAttemptAt = now;
diagnostics.enter("RUNTIME_PUBLICATION");
if (!publishRuntimeState()) {
inFlight.set(false);
return false;
@@ -130,9 +132,13 @@ public final class Ft546ShadowScheduler implements AutoCloseable {
Future<CycleAttempt> future = null;
try {
if (permanentlyStopped.get()) return false;
future = cycleExecutor.submit(() -> executeCycle(now, diagnostics));
CycleAttempt attempt = future.get(cycleDeadline.toMillis(),
TimeUnit.MILLISECONDS);
Instant overallDeadline = now.plus(cycleDeadline);
long remainingNanos = Math.min(cycleDeadline.toNanos() - (System.nanoTime() - startedNanos),
Duration.between(clock.instant(), overallDeadline).toNanos());
if (remainingNanos <= 0L) throw new TimeoutException("runtime publication exhausted cycle budget");
diagnostics.enter("QUEUED");
future = cycleExecutor.submit(() -> executeCycle(overallDeadline, diagnostics));
CycleAttempt attempt = future.get(remainingNanos, TimeUnit.NANOSECONDS);
outcome = attempt.result.getStatusCode();
statusReport.recordCycleDiagnostics(diagnostics.finish(outcome, null));
if ("DEGRADED".equals(attempt.result.getStatusCode())) {
@@ -212,32 +218,37 @@ public final class Ft546ShadowScheduler implements AutoCloseable {
}
}
private CycleAttempt executeCycle(Instant startedAt, ShadowCycleDiagnostics diagnostics) {
private CycleAttempt executeCycle(Instant overallDeadline, ShadowCycleDiagnostics diagnostics) {
cycleWorker = Thread.currentThread();
activeStage = CycleStage.SOURCE_READ;
diagnostics.enter("SOURCE_READ");
ShadowFactBatch batch = factReader.read(cursor,
startedAt.plus(sourceReadDeadline));
Instant sourceDeadline = clock.instant().plus(sourceReadDeadline);
if (sourceDeadline.isAfter(overallDeadline)) sourceDeadline = overallDeadline;
ShadowFactBatch batch = factReader.read(cursor, sourceDeadline);
activeStage = CycleStage.SHADOW_CYCLE;
return new CycleAttempt(batch,
shadowCycle.run(batch, clock.instant(), diagnostics));
}
private boolean handleFailure(Throwable failure) {
shadowCycle.invalidatePendingComparisons();
if (failure instanceof ShadowFactReadException) {
ShadowFactReadException readFailure =
(ShadowFactReadException) failure;
String timeoutReason = sourceTimeoutReason(readFailure);
if (timeoutReason != null) {
return stopPermanently(timeoutReason, readFailure,
CycleStage.SOURCE_READ);
if ("SOURCE_TRANSACTION_FAILED".equals(readFailure.getReasonCode())
|| "SOURCE_CYCLE_ALREADY_IN_FLIGHT".equals(readFailure.getReasonCode())) {
return stopPermanently(readFailure.getReasonCode(), readFailure, CycleStage.SOURCE_READ);
}
// ExecutionException is delivered only after the reader and its transaction have unwound.
// A watchdog timeout still stops permanently because cancellation does not prove termination.
String timeoutReason = sourceTimeoutReason(readFailure);
String reason = timeoutReason == null ? readFailure.getReasonCode() : timeoutReason;
Instant failedAt = clock.instant();
statusReport.markDegraded(readFailure.getReasonCode(), failedAt);
nextAttemptAt = failedAt.plus(readFailure.getRetryAfter());
Duration backoff = readFailure.getRetryAfter().isZero() ? CRASH_BACKOFF : readFailure.getRetryAfter();
statusReport.markDegraded(reason, failedAt);
nextAttemptAt = failedAt.plus(backoff);
LOGGER.warn("FT546 shadow source read failed: {}; retry after {} ms",
readFailure.getReasonCode(),
readFailure.getRetryAfter().toMillis());
reason, backoff.toMillis());
if (readFailure.getCause() != null) {
LOGGER.warn("FT546 shadow source read root cause",
readFailure.getCause());
@@ -278,6 +289,7 @@ public final class Ft546ShadowScheduler implements AutoCloseable {
private boolean stopPermanently(String reasonCode, Exception failure,
CycleStage stage) {
shadowCycle.invalidatePendingComparisons();
permanentlyStopped.set(true);
nextAttemptAt = Instant.EPOCH;
statusReport.markPermanentlyDegraded(reasonCode, clock.instant());