feat(ft546): persist shadow comparison evidence
This commit is contained in:
@@ -5,6 +5,12 @@ import com.ruoyi.ft546.protocol.Ft546SrgvCommandIntent;
|
||||
import com.ruoyi.shadow.facts.ShadowFactBatch;
|
||||
import com.ruoyi.shadow.facts.ShadowFactBatchValidator;
|
||||
import com.ruoyi.shadow.facts.ShadowFactValidation;
|
||||
import com.ruoyi.shadow.evidence.ShadowCycleEvidence;
|
||||
import com.ruoyi.shadow.evidence.ShadowCycleEvidenceCollector;
|
||||
import com.ruoyi.shadow.evidence.ShadowEvidenceException;
|
||||
import com.ruoyi.shadow.evidence.ShadowEvidenceRepository;
|
||||
import com.ruoyi.shadow.evidence.ShadowEvidenceStorageHealth;
|
||||
import com.ruoyi.shadow.startup.ShadowStartupReport;
|
||||
import com.ruoyi.wcs.application.ReadOnlyDispatchCycle;
|
||||
import com.ruoyi.wcs.application.ShadowDispatchResult;
|
||||
import com.ruoyi.wcs.application.dispatch.DispatchSnapshot;
|
||||
@@ -12,6 +18,7 @@ import com.ruoyi.wcs.application.dispatch.DispatchSnapshot;
|
||||
import java.time.Instant;
|
||||
import java.util.List;
|
||||
import java.util.Objects;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
import java.util.function.Function;
|
||||
|
||||
public final class Ft546ShadowCycle {
|
||||
@@ -19,6 +26,12 @@ public final class Ft546ShadowCycle {
|
||||
private final Function<ShadowFactBatch, DispatchSnapshot> assembler;
|
||||
private final ReadOnlyDispatchCycle<Ft546SrgvCommandIntent> dispatchCycle;
|
||||
private final RecordingFt546CommandGateway commandGateway;
|
||||
private final ShadowEvidenceStorageHealth storageHealth;
|
||||
private final ShadowEvidenceRepository evidenceRepository;
|
||||
private final ShadowCycleEvidenceCollector evidenceCollector;
|
||||
private final ShadowStartupReport statusReport;
|
||||
private final AtomicReference<String> degradedReason =
|
||||
new AtomicReference<String>();
|
||||
|
||||
public Ft546ShadowCycle(
|
||||
ShadowFactBatchValidator validator,
|
||||
@@ -31,9 +44,136 @@ public final class Ft546ShadowCycle {
|
||||
"dispatchCycle");
|
||||
this.commandGateway = Objects.requireNonNull(commandGateway,
|
||||
"commandGateway");
|
||||
this.storageHealth = null;
|
||||
this.evidenceRepository = null;
|
||||
this.evidenceCollector = null;
|
||||
this.statusReport = null;
|
||||
}
|
||||
|
||||
public Ft546ShadowCycle(
|
||||
ShadowFactBatchValidator validator,
|
||||
Function<ShadowFactBatch, DispatchSnapshot> assembler,
|
||||
ReadOnlyDispatchCycle<Ft546SrgvCommandIntent> dispatchCycle,
|
||||
RecordingFt546CommandGateway commandGateway,
|
||||
ShadowEvidenceStorageHealth storageHealth,
|
||||
ShadowEvidenceRepository evidenceRepository,
|
||||
ShadowCycleEvidenceCollector evidenceCollector,
|
||||
ShadowStartupReport statusReport) {
|
||||
this.validator = Objects.requireNonNull(validator, "validator");
|
||||
this.assembler = Objects.requireNonNull(assembler, "assembler");
|
||||
this.dispatchCycle = Objects.requireNonNull(dispatchCycle,
|
||||
"dispatchCycle");
|
||||
this.commandGateway = Objects.requireNonNull(commandGateway,
|
||||
"commandGateway");
|
||||
this.storageHealth = Objects.requireNonNull(storageHealth,
|
||||
"storageHealth");
|
||||
this.evidenceRepository = Objects.requireNonNull(evidenceRepository,
|
||||
"evidenceRepository");
|
||||
this.evidenceCollector = Objects.requireNonNull(evidenceCollector,
|
||||
"evidenceCollector");
|
||||
this.statusReport = Objects.requireNonNull(statusReport,
|
||||
"statusReport");
|
||||
}
|
||||
|
||||
public ShadowCycleResult run(ShadowFactBatch batch, Instant now) {
|
||||
if (evidenceRepository == null) {
|
||||
return runWithoutEvidence(batch, now);
|
||||
}
|
||||
String stoppedReason = degradedReason.get();
|
||||
if (stoppedReason != null) {
|
||||
return ShadowCycleResult.degraded(stoppedReason);
|
||||
}
|
||||
ShadowEvidenceStorageHealth.Check health = storageHealth.verify();
|
||||
if (!health.isAvailable()) {
|
||||
return degrade(now);
|
||||
}
|
||||
try {
|
||||
ShadowCycleResult result = evidenceRepository.inTransaction(
|
||||
() -> runAndPersist(batch, now));
|
||||
commandGateway.rememberAll(result.getEncodedCommands());
|
||||
Long lag = batch.getDatabaseCapturedAt() == null ? null
|
||||
: Long.valueOf(Math.max(0L,
|
||||
java.time.Duration.between(
|
||||
batch.getDatabaseCapturedAt(), now).toMillis()));
|
||||
statusReport.markReady(now, lag);
|
||||
return result;
|
||||
} catch (RuntimeException evidenceFailure) {
|
||||
return degrade(failureReason(evidenceFailure), now);
|
||||
}
|
||||
}
|
||||
|
||||
private ShadowCycleResult runAndPersist(ShadowFactBatch batch,
|
||||
Instant now) {
|
||||
ShadowFactValidation validation = validator.validate(batch, now);
|
||||
if (validation.getStatus()
|
||||
!= ShadowFactValidation.Status.COMPARABLE) {
|
||||
return persistInsufficient(batch, validation.getReasonCode(), now);
|
||||
}
|
||||
final DispatchSnapshot snapshot;
|
||||
try {
|
||||
snapshot = assembler.apply(batch);
|
||||
} catch (RuntimeException failure) {
|
||||
return persistInsufficient(batch,
|
||||
"INVALID_KERNEL_FACTS:" + message(failure), now);
|
||||
}
|
||||
ShadowDispatchResult<Ft546SrgvCommandIntent> result;
|
||||
try {
|
||||
result = dispatchCycle.run(snapshot);
|
||||
} catch (IllegalArgumentException failure) {
|
||||
return persistInsufficient(batch,
|
||||
"INVALID_COMMAND_FACTS:" + message(failure), now);
|
||||
}
|
||||
if (result.getDecisions().isRetrySnapshotChanged()) {
|
||||
List<ShadowCycleEvidence> evidence = evidenceCollector.insufficient(
|
||||
batch, "SHADOW_TRAFFIC_STATE_CHANGED", now);
|
||||
evidenceRepository.saveAll(evidence);
|
||||
return ShadowCycleResult.retry(result.getDecisions());
|
||||
}
|
||||
final List<Ft546EncodedCommand> encoded;
|
||||
try {
|
||||
encoded = commandGateway.previewAll(result.getIntents());
|
||||
} catch (RuntimeException failure) {
|
||||
return persistInsufficient(batch,
|
||||
"INVALID_COMMAND_FACTS:" + message(failure), now);
|
||||
}
|
||||
evidenceRepository.saveAll(evidenceCollector.collect(batch,
|
||||
result.getDecisions(), encoded, now));
|
||||
return ShadowCycleResult.comparable(result.getDecisions(),
|
||||
result.getIntents(), encoded);
|
||||
}
|
||||
|
||||
private ShadowCycleResult persistInsufficient(
|
||||
ShadowFactBatch batch, String reasonCode, Instant now) {
|
||||
evidenceRepository.saveAll(evidenceCollector.insufficient(batch,
|
||||
reasonCode, now));
|
||||
return ShadowCycleResult.insufficient(reasonCode);
|
||||
}
|
||||
|
||||
private ShadowCycleResult degrade(Instant now) {
|
||||
return degrade("EVIDENCE_STORAGE_UNAVAILABLE", now);
|
||||
}
|
||||
|
||||
private ShadowCycleResult degrade(String reasonCode, Instant now) {
|
||||
degradedReason.compareAndSet(null, reasonCode);
|
||||
String stoppedReason = degradedReason.get();
|
||||
statusReport.markDegraded(stoppedReason, now);
|
||||
return ShadowCycleResult.degraded(stoppedReason);
|
||||
}
|
||||
|
||||
private static String failureReason(RuntimeException failure) {
|
||||
if (failure instanceof ShadowEvidenceException) {
|
||||
String reasonCode = ((ShadowEvidenceException) failure)
|
||||
.getReasonCode();
|
||||
if ("EVIDENCE_KEY_COLLISION".equals(reasonCode)
|
||||
|| "EVIDENCE_RUN_COLLISION".equals(reasonCode)) {
|
||||
return reasonCode;
|
||||
}
|
||||
}
|
||||
return "EVIDENCE_STORAGE_UNAVAILABLE";
|
||||
}
|
||||
|
||||
private ShadowCycleResult runWithoutEvidence(ShadowFactBatch batch,
|
||||
Instant now) {
|
||||
ShadowFactValidation validation = validator.validate(batch, now);
|
||||
if (validation.getStatus()
|
||||
!= ShadowFactValidation.Status.COMPARABLE) {
|
||||
|
||||
@@ -44,16 +44,28 @@ public final class RecordingFt546CommandGateway {
|
||||
|
||||
public synchronized List<Ft546EncodedCommand> recordAll(
|
||||
List<Ft546SrgvCommandIntent> intents) {
|
||||
List<Ft546EncodedCommand> encoded = previewAll(intents);
|
||||
rememberAll(encoded);
|
||||
return encoded;
|
||||
}
|
||||
|
||||
public synchronized List<Ft546EncodedCommand> previewAll(
|
||||
List<Ft546SrgvCommandIntent> intents) {
|
||||
Objects.requireNonNull(intents, "intents");
|
||||
List<Ft546EncodedCommand> encoded =
|
||||
new ArrayList<Ft546EncodedCommand>(intents.size());
|
||||
for (Ft546SrgvCommandIntent intent : intents) {
|
||||
encoded.add(preview(intent));
|
||||
}
|
||||
recorded.addAll(encoded);
|
||||
return Collections.unmodifiableList(encoded);
|
||||
}
|
||||
|
||||
public synchronized void rememberAll(
|
||||
List<Ft546EncodedCommand> encodedCommands) {
|
||||
recorded.addAll(new ArrayList<Ft546EncodedCommand>(
|
||||
Objects.requireNonNull(encodedCommands, "encodedCommands")));
|
||||
}
|
||||
|
||||
public synchronized List<Ft546EncodedCommand> getRecorded() {
|
||||
return Collections.unmodifiableList(
|
||||
new ArrayList<Ft546EncodedCommand>(recorded));
|
||||
|
||||
@@ -47,6 +47,12 @@ public final class ShadowCycleResult {
|
||||
encodedCommands);
|
||||
}
|
||||
|
||||
public static ShadowCycleResult degraded(String reasonCode) {
|
||||
return new ShadowCycleResult("DEGRADED", reasonCode, null,
|
||||
Collections.<Ft546SrgvCommandIntent>emptyList(),
|
||||
Collections.<Ft546EncodedCommand>emptyList());
|
||||
}
|
||||
|
||||
public String getStatusCode() { return statusCode; }
|
||||
public String getReasonCode() { return reasonCode; }
|
||||
public DispatchCycleResult getDecisions() { return decisions; }
|
||||
|
||||
Reference in New Issue
Block a user