diff --git a/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowObservationGateway.java b/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowObservationGateway.java index 15a7b72..22da546 100644 --- a/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowObservationGateway.java +++ b/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowObservationGateway.java @@ -283,6 +283,17 @@ public class ShadowObservationGateway { metrics.increment(ShadowSkipReason.CONCURRENT); } catch (Exception shadowFailure) { coordinator.failRefresh(ticket, failureDescription(failureStage, shadowFailure)); + if (isSourceChanged(shadowFailure)) { + // SQL 51025 is the stored procedure's optimistic-concurrency guard: + // source rows changed while the snapshot was being built. Treat it as + // a stale refresh and defer the next attempt; retrying immediately only + // creates another collision and incorrectly reports an application error. + metrics.increment(ShadowSkipReason.STALE); + recoveryNotBefore.set(clock.instant().plus(RECOVERY_COOLDOWN)); + LOG.info("shadow_refresh_stale warehouse={} snapshotGeneration={} sqlCode=51025", + warehouseId, generation); + return; + } metrics.recordError(failureStage); warnFailure("CONTROL", generation, failureStage, shadowFailure, startedNanos); } @@ -290,6 +301,21 @@ public class ShadowObservationGateway { recoveryNotBefore.set(clock.instant().plus(RECOVERY_COOLDOWN)); } + private boolean isSourceChanged(Throwable failure) { + Throwable cause = failure; + for (int depth = 0; depth < 8 && cause != null; depth++) { + if (cause instanceof SQLException sql && sql.getErrorCode() == 51025) { + return true; + } + Throwable next = cause.getCause(); + if (next == cause) { + break; + } + cause = next; + } + return false; + } + private Optional tryBeginRefreshAfterQuietWindow() throws InterruptedException { if (coordinator.activeLegacyCount() != 0) { diff --git a/ruoyi-admin/src/test/java/com/ruoyi/algorithm/shadow/ShadowObservationGatewayTest.java b/ruoyi-admin/src/test/java/com/ruoyi/algorithm/shadow/ShadowObservationGatewayTest.java index 66ea376..8eb2850 100644 --- a/ruoyi-admin/src/test/java/com/ruoyi/algorithm/shadow/ShadowObservationGatewayTest.java +++ b/ruoyi-admin/src/test/java/com/ruoyi/algorithm/shadow/ShadowObservationGatewayTest.java @@ -9,6 +9,7 @@ import java.time.Duration; import java.time.Instant; import java.time.ZoneId; import java.time.ZoneOffset; +import java.sql.SQLException; import java.util.ArrayList; import java.util.List; import java.util.concurrent.ArrayBlockingQueue; @@ -392,6 +393,24 @@ class ShadowObservationGatewayTest { assertThat(recovered.await(1, TimeUnit.SECONDS)).isTrue(); } + @Test + void sourceChangedRefreshIsClassifiedAsStaleAndDoesNotRetry() throws Exception { + ShadowPlanningEngine engine = mock(ShadowPlanningEngine.class); + ShadowInputMapper mapper = mock(ShadowInputMapper.class); + SQLException sourceChanged = new SQLException("source changed", "S0001", 51025); + when(engine.refreshSnapshot(anyLong(), eq("INSTANCE-1"))).thenThrow(new RuntimeException(sourceChanged)); + CountDownLatch staleMetric = metricLatch(mapper, "SKIP_STALE"); + ShadowObservationGateway gateway = gateway(engine, mapper, coordinator(), + new ShadowRuntimeMetrics(), Clock.fixed(NOW, ZoneOffset.UTC), delay -> { }); + + gateway.start(); + + assertThat(staleMetric.await(1, TimeUnit.SECONDS)).isTrue(); + verify(engine, timeout(1000).times(1)).refreshSnapshot(anyLong(), eq("INSTANCE-1")); + verify(mapper, never()).upsertRuntimeMetric(anyString(), eq("ERROR_REFRESH"), anyLong(), + anyLong(), anyLong(), anyLong(), any(Instant.class)); + } + private void assertOutboundUsesLease(boolean exactStock) throws Exception { ShadowPlanningEngine engine = mock(ShadowPlanningEngine.class); ShadowInputMapper mapper = mock(ShadowInputMapper.class);