zyr/FT546MWS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: observe legacy failures in shadow worker

Browse Source
This commit is contained in:
zgw
2026-09-03 16:32:31 +08:00
parent 87700246d8
commit ffd3cc04f3
3 changed files with 117 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowObservationGateway.java
View File

@@ -156,6 +156,7 @@ public class ShadowObservationGateway {
stage = ShadowFailureStage.LEGACY_WAIT;
InboundShadowCompletion.LegacyResult legacy = completion.awaitLegacyResult();
if (legacy.failure() != null) {
recordWorkerFailure(taskNo, admission, stage, legacy.failure(), startedNanos);
return;
}
if (!coordinator.isComparable(admission)) {
@@ -185,6 +186,7 @@ public class ShadowObservationGateway {
stage = ShadowFailureStage.LEGACY_WAIT;
OutboundShadowCompletion.LegacyResult legacy = completion.awaitLegacyResult();
if (legacy.failure() != null) {
recordWorkerFailure(stockoutDetailId, admission, stage, legacy.failure(), startedNanos);
return;
}
if (!coordinator.isComparable(admission)) {
@@ -334,7 +336,7 @@ public class ShadowObservationGateway {
private void recordWorkerFailure(String requestId,
ShadowSnapshotCoordinator.Admission admission, ShadowFailureStage stage,
Exception failure, long startedNanos) {
Throwable failure, long startedNanos) {
metrics.recordError(stage);
warnFailure(requestId, admission.snapshot().generation(), stage, failure, startedNanos);
}

45
ruoyi-admin/src/test/java/com/ruoyi/algorithm/shadow/ShadowObservationGatewayTest.java
View File

@@ -255,6 +255,51 @@ class ShadowObservationGatewayTest {
assertThat(coordinator.activeLegacyCount()).isZero();
}
@Test
void inboundLegacyFailureIsRecordedAtLegacyWaitStage() throws Exception {
ShadowPlanningEngine engine = mock(ShadowPlanningEngine.class);
ShadowInputMapper mapper = mock(ShadowInputMapper.class);
ShadowSnapshotCoordinator coordinator = coordinatorWithSnapshot();
when(engine.planInbound(SNAPSHOT, "TASK-FAIL", "TRAY-FAIL")).thenReturn(inboundObservation());
CountDownLatch waitMetric = metricLatch(mapper, "ERROR_LEGACY_WAIT");
stubSuccessfulRecovery(engine, mapper, 43L);
ShadowObservationGateway gateway = gateway(engine, mapper, coordinator,
new ShadowRuntimeMetrics(), Clock.fixed(NOW, ZoneOffset.UTC), delay -> { });
InboundShadowCompletion completion = gateway.observeInbound("TASK-FAIL", "TRAY-FAIL");
completion.fail(new IllegalStateException("business payload must not be logged"));
assertThat(waitMetric.await(1, TimeUnit.SECONDS)).isTrue();
assertThat(coordinator.activeLegacyCount()).isZero();
completion.fail(new IllegalStateException("duplicate"));
completion.complete("10", "SPACE-FAIL");
assertThat(coordinator.activeLegacyCount()).isZero();
}
@Test
void outboundLegacyFailureIsRecordedAtLegacyWaitStage() throws Exception {
ShadowPlanningEngine engine = mock(ShadowPlanningEngine.class);
ShadowInputMapper mapper = mock(ShadowInputMapper.class);
ShadowSnapshotCoordinator coordinator = coordinatorWithSnapshot();
OutboundShadowObservation observation = new OutboundShadowObservation(
true, 42L, "01", NOW, "ABC123", "DETAIL-FAIL", "READY",
"STOCK-FAIL", "SPACE-FAIL", "diagnostic");
when(engine.planProjectOutbound(SNAPSHOT, "DETAIL-FAIL")).thenReturn(observation);
CountDownLatch waitMetric = metricLatch(mapper, "ERROR_LEGACY_WAIT");
stubSuccessfulRecovery(engine, mapper, 43L);
ShadowObservationGateway gateway = gateway(engine, mapper, coordinator,
new ShadowRuntimeMetrics(), Clock.fixed(NOW, ZoneOffset.UTC), delay -> { });
OutboundShadowCompletion completion = gateway.observeProjectOutbound("DETAIL-FAIL");
completion.fail(new IllegalStateException("business payload must not be logged"));
assertThat(waitMetric.await(1, TimeUnit.SECONDS)).isTrue();
assertThat(coordinator.activeLegacyCount()).isZero();
completion.fail(new IllegalStateException("duplicate"));
completion.complete("10");
assertThat(coordinator.activeLegacyCount()).isZero();
}
@Test
void projectAndExactStockOutboundUseTheLeasedGeneration() throws Exception {
assertOutboundUsesLease(false);