1 Commits

Author SHA1 Message Date
zgw
63ed8f72b1 fix(shadow): classify source-change refresh as stale 2026-09-10 11:15:28 +08:00
2 changed files with 45 additions and 0 deletions

View File

@@ -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<ShadowSnapshotCoordinator.RefreshTicket> tryBeginRefreshAfterQuietWindow()
throws InterruptedException {
if (coordinator.activeLegacyCount() != 0) {

View File

@@ -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);