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

fix(shadow): repair field routing and harden evidence collection

Browse Source
This commit is contained in:
zgw
2026-09-07 12:42:17 +08:00
parent 9829b9c0fa
commit 414b3bf4a9
37 changed files with 1751 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -14,6 +14,7 @@ import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.util.Objects;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -22,6 +23,7 @@ import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
public final class Ft546ShadowScheduler implements AutoCloseable {
private static final Logger LOGGER =
@@ -39,11 +41,17 @@ public final class Ft546ShadowScheduler implements AutoCloseable {
private final Duration cycleDeadline;
private final Duration sourceReadDeadline;
private final ExecutorService cycleExecutor;
private final Consumer<Map<String, Object>> runtimeWriter;
private final Object runtimeLock = new Object();
private final AtomicBoolean closed = new AtomicBoolean();
private volatile Instant lastAttemptAt;
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;
private volatile Thread cycleWorker;
private long lastTimingLogNanos;
public Ft546ShadowScheduler(SqlServerShadowFactReader factReader,
Ft546ShadowCycle shadowCycle,
@@ -67,6 +75,21 @@ public final class Ft546ShadowScheduler implements AutoCloseable {
Clock clock,
Duration cycleDeadline,
Duration sourceReadDeadline) {
this(factReader, shadowCycle, statusReport, clock, cycleDeadline, sourceReadDeadline, state -> { });
}
public Ft546ShadowScheduler(SqlServerShadowFactReader factReader,
Ft546ShadowCycle shadowCycle, ShadowStartupReport statusReport,
Clock clock, Duration cycleDeadline,
Consumer<Map<String, Object>> runtimeWriter) {
this(factReader, shadowCycle, statusReport, clock, cycleDeadline,
shorter(cycleDeadline, SOURCE_READ_DEADLINE), runtimeWriter);
}
private Ft546ShadowScheduler(SqlServerShadowFactReader factReader,
Ft546ShadowCycle shadowCycle, ShadowStartupReport statusReport,
Clock clock, Duration cycleDeadline, Duration sourceReadDeadline,
Consumer<Map<String, Object>> runtimeWriter) {
this.factReader = Objects.requireNonNull(factReader, "factReader");
this.shadowCycle = Objects.requireNonNull(shadowCycle, "shadowCycle");
this.statusReport = Objects.requireNonNull(statusReport,
@@ -74,6 +97,7 @@ public final class Ft546ShadowScheduler implements AutoCloseable {
this.clock = Objects.requireNonNull(clock, "clock");
this.cycleDeadline = requirePositive(cycleDeadline);
this.sourceReadDeadline = requirePositive(sourceReadDeadline);
this.runtimeWriter = Objects.requireNonNull(runtimeWriter, "runtimeWriter");
this.cycleExecutor = Executors.newSingleThreadExecutor(
shadowCycleThreadFactory());
}
@@ -82,7 +106,11 @@ public final class Ft546ShadowScheduler implements AutoCloseable {
initialDelayString = "${shadow.poll.initial-delay-ms:5000}",
fixedDelayString = "${shadow.poll.fixed-delay-ms:5000}")
public void runScheduledCycle() {
runOnce();
if (permanentlyStopped.get() || clock.instant().isBefore(nextAttemptAt)) {
publishRuntimeState();
} else {
runOnce();
}
}
public boolean runOnce() {
@@ -91,11 +119,22 @@ public final class Ft546ShadowScheduler implements AutoCloseable {
|| !inFlight.compareAndSet(false, true)) {
return false;
}
Future<CycleAttempt> future = cycleExecutor.submit(
() -> executeCycle(now));
ShadowCycleDiagnostics diagnostics = new ShadowCycleDiagnostics();
lastAttemptAt = now;
if (!publishRuntimeState()) {
inFlight.set(false);
return false;
}
cycleWorker = null;
String outcome = "FAILED";
Future<CycleAttempt> future = null;
try {
if (permanentlyStopped.get()) return false;
future = cycleExecutor.submit(() -> executeCycle(now, diagnostics));
CycleAttempt attempt = future.get(cycleDeadline.toMillis(),
TimeUnit.MILLISECONDS);
outcome = attempt.result.getStatusCode();
statusReport.recordCycleDiagnostics(diagnostics.finish(outcome, null));
if ("DEGRADED".equals(attempt.result.getStatusCode())) {
permanentlyStopped.set(true);
statusReport.markPermanentlyDegraded(
@@ -109,20 +148,34 @@ public final class Ft546ShadowScheduler implements AutoCloseable {
return true;
} catch (TimeoutException timeout) {
CycleStage timedOutStage = activeStage;
future.cancel(true);
statusReport.recordCycleDiagnostics(diagnostics.cancel("TIMEOUT", cycleWorker));
if (future != null) future.cancel(true);
return stopPermanently("SHADOW_CYCLE_TIMEOUT", timeout,
timedOutStage);
} catch (InterruptedException interrupted) {
CycleStage interruptedStage = activeStage;
future.cancel(true);
statusReport.recordCycleDiagnostics(diagnostics.cancel("INTERRUPTED", cycleWorker));
if (future != null) future.cancel(true);
Thread.currentThread().interrupt();
return stopPermanently("SHADOW_CYCLE_INTERRUPTED", interrupted,
interruptedStage);
} catch (ExecutionException wrapped) {
statusReport.recordCycleDiagnostics(diagnostics.finish("FAILED", null));
return handleFailure(wrapped.getCause());
} catch (java.util.concurrent.RejectedExecutionException shutdown) {
return stopPermanently("SHADOW_EXECUTOR_STOPPED", shutdown, activeStage);
} finally {
Map<String, Object> 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;
}
inFlight.set(false);
activeStage = CycleStage.IDLE;
publishRuntimeState();
}
}
@@ -137,17 +190,37 @@ public final class Ft546ShadowScheduler implements AutoCloseable {
@Override
public void close() {
if (!closed.compareAndSet(false, true)) return;
permanentlyStopped.set(true);
statusReport.markStopped(clock.instant());
cycleExecutor.shutdownNow();
publishRuntimeState();
}
private CycleAttempt executeCycle(Instant startedAt) {
private boolean publishRuntimeState() {
synchronized (runtimeLock) {
try {
runtimeWriter.accept(statusReport.snapshotRuntimeState(
clock.instant(), lastAttemptAt, permanentlyStopped.get()));
return true;
} catch (RuntimeException failure) {
permanentlyStopped.set(true);
statusReport.markPermanentlyDegraded("EVIDENCE_RUNTIME_WRITE_FAILED", clock.instant());
LOGGER.error("FT546 shadow runtime state unavailable; collection stopped", failure);
return false;
}
}
}
private CycleAttempt executeCycle(Instant startedAt, ShadowCycleDiagnostics diagnostics) {
cycleWorker = Thread.currentThread();
activeStage = CycleStage.SOURCE_READ;
diagnostics.enter("SOURCE_READ");
ShadowFactBatch batch = factReader.read(cursor,
startedAt.plus(sourceReadDeadline));
activeStage = CycleStage.SHADOW_CYCLE;
return new CycleAttempt(batch,
shadowCycle.run(batch, clock.instant()));
shadowCycle.run(batch, clock.instant(), diagnostics));
}
private boolean handleFailure(Throwable failure) {
@@ -208,8 +281,8 @@ public final class Ft546ShadowScheduler implements AutoCloseable {
permanentlyStopped.set(true);
nextAttemptAt = Instant.EPOCH;
statusReport.markPermanentlyDegraded(reasonCode, clock.instant());
LOGGER.error("FT546 shadow collection stopped: {}; stage={}",
reasonCode, stage, failure);
LOGGER.error("FT546 shadow collection stopped: {}; stage={}; diagnostics={}",
reasonCode, stage, statusReport.getCycleDiagnostics(), failure);
return false;
}