From b2b3f00f6b24144d1748bcfa163dcbbce090c107 Mon Sep 17 00:00:00 2001 From: dao-jun Date: Mon, 2 Nov 2026 00:17:23 +0800 Subject: [PATCH 1/3] fix: gate stale read when snapshots installing --- .../ratis/server/impl/ReadRequestHandler.java | 6 + .../impl/SnapshotInstallationHandler.java | 6 + ...stStaleReadDuringSnapshotInstallation.java | 105 ++++++++++++++++++ 3 files changed, 117 insertions(+) create mode 100644 ratis-server/src/test/java/org/apache/ratis/server/impl/TestStaleReadDuringSnapshotInstallation.java diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/ReadRequestHandler.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/ReadRequestHandler.java index 9b961c5773..a8f3f8a04f 100644 --- a/ratis-server/src/main/java/org/apache/ratis/server/impl/ReadRequestHandler.java +++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/ReadRequestHandler.java @@ -81,6 +81,12 @@ RaftGroupMemberId getMemberId() { } CompletableFuture staleReadAsync(RaftClientRequest request) { + final long installSnapshot = server.getSnapshotInstallationHandler().getInProgressInstallSnapshotIndex(); + final ReadException ex = getReadException("stale read", installSnapshot, false); + if (ex != null) { + return CompletableFuture.completedFuture( + server.newExceptionReply(request, new StateMachineException(getMemberId(), ex))); + } final long minIndex = request.getType().getStaleRead().getMinIndex(); final long commitIndex = state.getLog().getLastCommittedIndex(); LOG.debug("{}: minIndex={}, commitIndex={} from {}", getMemberId(), minIndex, commitIndex, request.getClientId()); diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/SnapshotInstallationHandler.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/SnapshotInstallationHandler.java index d34e84c3fb..5e58c76074 100644 --- a/ratis-server/src/main/java/org/apache/ratis/server/impl/SnapshotInstallationHandler.java +++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/SnapshotInstallationHandler.java @@ -43,6 +43,7 @@ import org.apache.ratis.util.CodeInjectionForTesting; import org.apache.ratis.util.LifeCycle; import org.apache.ratis.util.Preconditions; +import org.apache.ratis.thirdparty.com.google.common.annotations.VisibleForTesting; import org.apache.ratis.util.ProtoUtils; import org.apache.ratis.util.Timestamp; import org.slf4j.Logger; @@ -106,6 +107,11 @@ long getInProgressInstallSnapshotIndex() { return inProgressInstallSnapshotIndex.get(); } + @VisibleForTesting + void setInProgressInstallSnapshotIndex(long index) { + inProgressInstallSnapshotIndex.set(index); + } + InstallSnapshotReplyProto installSnapshot(InstallSnapshotRequestProto request) throws IOException { BatchLogger.print(BatchLogKey.INSTALL_SNAPSHOT_REQUEST, getMemberId(), suffix -> LOG.info("{}: receive installSnapshot: {} {}", diff --git a/ratis-server/src/test/java/org/apache/ratis/server/impl/TestStaleReadDuringSnapshotInstallation.java b/ratis-server/src/test/java/org/apache/ratis/server/impl/TestStaleReadDuringSnapshotInstallation.java new file mode 100644 index 0000000000..dcc650bc0a --- /dev/null +++ b/ratis-server/src/test/java/org/apache/ratis/server/impl/TestStaleReadDuringSnapshotInstallation.java @@ -0,0 +1,105 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.ratis.server.impl; + +import org.apache.ratis.RaftTestUtil; +import org.apache.ratis.client.RaftClient; +import org.apache.ratis.conf.RaftProperties; +import org.apache.ratis.protocol.Message; +import org.apache.ratis.protocol.RaftClientReply; +import org.apache.ratis.protocol.exceptions.StateMachineException; +import org.apache.ratis.server.RaftServer; +import org.apache.ratis.server.raftlog.RaftLog; +import org.apache.ratis.server.simulation.MiniRaftClusterWithSimulatedRpc; +import org.apache.ratis.statemachine.StateMachine; +import org.apache.ratis.statemachine.impl.SimpleStateMachine4Testing; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import java.io.IOException; + +/** + * Test stale reads failing while a snapshot installation is in progress: the follower serving the + * stale read is exactly the node whose state machine is paused/reinitialized by the installation, + * so the read must be rejected with a retriable error instead of querying a paused state machine + * (see {@link ReadRequestHandler#staleReadAsync}). + */ +public class TestStaleReadDuringSnapshotInstallation { + private static final int NUM_SERVERS = 3; + + private final MiniRaftClusterWithSimulatedRpc cluster = MiniRaftClusterWithSimulatedRpc + .FACTORY.newCluster(NUM_SERVERS, getProperties()); + + private static RaftProperties getProperties() { + final RaftProperties properties = new RaftProperties(); + properties.setClass(MiniRaftCluster.STATEMACHINE_CLASS_KEY, + SimpleStateMachine4Testing.class, StateMachine.class); + return properties; + } + + @BeforeEach + public void setup() throws IOException { + cluster.start(); + } + + @AfterEach + public void tearDown() { + cluster.shutdown(); + } + + @Test + @Timeout(120) + public void testStaleReadRejectedWhileSnapshotInstallationInProgress() throws Exception { + RaftTestUtil.waitForLeader(cluster); + + try (final RaftClient client = cluster.createClient()) { + for (int i = 0; i < 5; i++) { + Assertions.assertTrue(client.io().send(Message.valueOf("m" + i)).isSuccess()); + } + + final RaftServer.Division follower = cluster.getFollowers().get(0); + final SnapshotInstallationHandler handler = + ((RaftServerImpl) follower).getSnapshotInstallationHandler(); + final Message message = Message.valueOf("m4"); + + // Regression guard: a stale read on a healthy follower still succeeds. + final RaftClientReply before = client.io().sendStaleRead(message, 0, follower.getId()); + Assertions.assertTrue(before.isSuccess()); + + // Simulate a snapshot installation in progress on the follower. + handler.setInProgressInstallSnapshotIndex(1); + try { + final IOException e = Assertions.assertThrows(IOException.class, + () -> client.io().sendStaleRead(message, 0, follower.getId())); + // The gate rejects with a retriable StateMachineException, not a success and not a crash. + Assertions.assertTrue(e instanceof StateMachineException, () -> "Unexpected exception: " + e); + Assertions.assertTrue(e.getMessage().contains("stale read"), () -> e.getMessage()); + Assertions.assertTrue(e.getMessage().contains("in progress"), () -> e.getMessage()); + } finally { + handler.setInProgressInstallSnapshotIndex(RaftLog.INVALID_LOG_INDEX); + } + + // After the (simulated) installation completed: stale reads work again. + final RaftClientReply after = client.io().sendStaleRead(message, 0, follower.getId()); + Assertions.assertTrue(after.isSuccess()); + } + } +} From 8a76358858eee1607cd76fabd300cffc2de0c17e Mon Sep 17 00:00:00 2001 From: dao-jun Date: Mon, 2 Nov 2026 04:02:58 +0800 Subject: [PATCH 2/3] fix tests --- .../TestStaleReadDuringSnapshotInstallation.java | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/ratis-server/src/test/java/org/apache/ratis/server/impl/TestStaleReadDuringSnapshotInstallation.java b/ratis-server/src/test/java/org/apache/ratis/server/impl/TestStaleReadDuringSnapshotInstallation.java index dcc650bc0a..c713b9cc24 100644 --- a/ratis-server/src/test/java/org/apache/ratis/server/impl/TestStaleReadDuringSnapshotInstallation.java +++ b/ratis-server/src/test/java/org/apache/ratis/server/impl/TestStaleReadDuringSnapshotInstallation.java @@ -80,15 +80,25 @@ public void testStaleReadRejectedWhileSnapshotInstallationInProgress() throws Ex ((RaftServerImpl) follower).getSnapshotInstallationHandler(); final Message message = Message.valueOf("m4"); + // send() only guarantees that the leader has committed; the follower applies asynchronously. + // Wait until the follower has applied m4, otherwise a stale read may hit the state machine + // before the queried entry is applied (see SimpleStateMachine4Testing#query). + final long m4Index = cluster.getLeader().getRaftLog().getLastCommittedIndex(); + for (int i = 0; follower.getInfo().getLastAppliedIndex() < m4Index; i++) { + Assertions.assertTrue(i < 100, () -> follower.getId() + ": fails to catch up, lastAppliedIndex=" + + follower.getInfo().getLastAppliedIndex() + " < " + m4Index); + Thread.sleep(100); + } + // Regression guard: a stale read on a healthy follower still succeeds. - final RaftClientReply before = client.io().sendStaleRead(message, 0, follower.getId()); + final RaftClientReply before = client.io().sendStaleRead(message, m4Index, follower.getId()); Assertions.assertTrue(before.isSuccess()); // Simulate a snapshot installation in progress on the follower. handler.setInProgressInstallSnapshotIndex(1); try { final IOException e = Assertions.assertThrows(IOException.class, - () -> client.io().sendStaleRead(message, 0, follower.getId())); + () -> client.io().sendStaleRead(message, m4Index, follower.getId())); // The gate rejects with a retriable StateMachineException, not a success and not a crash. Assertions.assertTrue(e instanceof StateMachineException, () -> "Unexpected exception: " + e); Assertions.assertTrue(e.getMessage().contains("stale read"), () -> e.getMessage()); @@ -98,7 +108,7 @@ public void testStaleReadRejectedWhileSnapshotInstallationInProgress() throws Ex } // After the (simulated) installation completed: stale reads work again. - final RaftClientReply after = client.io().sendStaleRead(message, 0, follower.getId()); + final RaftClientReply after = client.io().sendStaleRead(message, m4Index, follower.getId()); Assertions.assertTrue(after.isSuccess()); } } From 2e2f4d0e690b0cc0ff9e0d7803537f45c1065875 Mon Sep 17 00:00:00 2001 From: dao-jun Date: Fri, 6 Nov 2026 12:01:28 +0800 Subject: [PATCH 3/3] Address review: reuse installing-snapshot gate for stale read, cover chunk-based install - Add op parameter to getReadExceptionIfInstallingSnapshot and drop the hardcoded readIndex from the ReadException message (szetszwo) - Set inProgressInstallSnapshotIndex in checkAndInstallSnapshot around the pause/finalize/reload window so the gate covers both install paths (szetszwo, agreed by amaliujia) - Reply with the ReadException directly instead of wrapping it in a StateMachineException, which must come from the state machine (szetszwo) - Test the withholding semantics: with the gate held, a blocking stale read keeps retrying the retriable ReadException instead of returning data from a paused state machine, and completes after the installation finishes --- .../ratis/server/impl/ReadRequestHandler.java | 25 +++++++++---------- .../impl/SnapshotInstallationHandler.java | 19 ++++++++++---- ...stStaleReadDuringSnapshotInstallation.java | 23 +++++++++-------- 3 files changed, 39 insertions(+), 28 deletions(-) diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/ReadRequestHandler.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/ReadRequestHandler.java index a8f3f8a04f..aaaa92b265 100644 --- a/ratis-server/src/main/java/org/apache/ratis/server/impl/ReadRequestHandler.java +++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/ReadRequestHandler.java @@ -81,11 +81,9 @@ RaftGroupMemberId getMemberId() { } CompletableFuture staleReadAsync(RaftClientRequest request) { - final long installSnapshot = server.getSnapshotInstallationHandler().getInProgressInstallSnapshotIndex(); - final ReadException ex = getReadException("stale read", installSnapshot, false); + final ReadException ex = getReadExceptionIfInstallingSnapshot("stale read"); if (ex != null) { - return CompletableFuture.completedFuture( - server.newExceptionReply(request, new StateMachineException(getMemberId(), ex))); + return CompletableFuture.completedFuture(server.newExceptionReply(request, ex)); } final long minIndex = request.getType().getStaleRead().getMinIndex(); final long commitIndex = state.getLog().getLastCommittedIndex(); @@ -100,19 +98,20 @@ CompletableFuture staleReadAsync(RaftClientRequest request) { } ReadException getReadException(String op, long installSnapshot, boolean started) { - return installSnapshot == RaftLog.INVALID_LOG_INDEX ? null : new ReadException(getMemberId() + ": Failed to " + op - + " readIndex as snapshot (" + installSnapshot + ") installation is " + (started ? "started" : "in progress")); + return installSnapshot == RaftLog.INVALID_LOG_INDEX ? null : new ReadException(getMemberId() + ": Failed " + op + + " as snapshot (" + installSnapshot + ") installation is " + (started ? "started" : "in progress")); } - private ReadException getReadExceptionIfInstallingSnapshot() { - return getReadException("add", server.getSnapshotInstallationHandler().getInProgressInstallSnapshotIndex(), false); + private ReadException getReadExceptionIfInstallingSnapshot(String op) { + final long installSnapshotIndex = server.getSnapshotInstallationHandler().getInProgressInstallSnapshotIndex(); + return getReadException(op, installSnapshotIndex, false); } private CompletableFuture sendReadIndexAsync( ClientId clientId, ReadRequestTypeProto readRequestType) { - final long installSnapshot = server.getSnapshotInstallationHandler().getInProgressInstallSnapshotIndex(); - if (installSnapshot != RaftLog.INVALID_LOG_INDEX) { - return JavaUtils.completeExceptionally(getReadException("get", installSnapshot, false)); + final ReadException readException = getReadExceptionIfInstallingSnapshot("to send readIndex request"); + if (readException != null) { + return JavaUtils.completeExceptionally(readException); } final RaftPeerId leaderId = server.getInfo().getLeaderId(); if (leaderId == null) { @@ -160,7 +159,7 @@ CompletableFuture readOnlyAsync(ClientId clientId, ReadRequestTypeProto final LeaderStateImpl leader = server.getRole().getLeaderState().orElse(null); return getReadIndex(clientId, readRequestType, leader) .thenCompose(readIndex -> state.getReadRequests().waitToAdvance(readIndex, - this::getReadExceptionIfInstallingSnapshot)) + () -> getReadExceptionIfInstallingSnapshot("local read"))) .thenCompose(readIndex -> stateMachine.query(queryMessage)); } else { throw new IllegalStateException("Unexpected read option: " + readOption); @@ -183,7 +182,7 @@ CompletableFuture readAsync(RaftClientRequest request) { return replyFuture .thenCompose(readIndex -> state.getReadRequests().waitToAdvance(readIndex, - this::getReadExceptionIfInstallingSnapshot)) + () -> getReadExceptionIfInstallingSnapshot("read"))) .thenCompose(readIndex -> isDummyRead(request) ? CompletableFuture.completedFuture(server.newSuccessReply(request)) : queryStateMachine(request)) .exceptionally(e -> readException2Reply(request, e)); diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/SnapshotInstallationHandler.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/SnapshotInstallationHandler.java index 5e58c76074..5556322663 100644 --- a/ratis-server/src/main/java/org/apache/ratis/server/impl/SnapshotInstallationHandler.java +++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/SnapshotInstallationHandler.java @@ -240,10 +240,19 @@ private CompletableFuture checkAndInstallSnapshot(Ins // update the committed index // re-load the state machine if this is the last chunk if (snapshotChunkRequest.getDone()) { - stateMachine.pause(); // pause the SM right before publishing the snapshot atomically - snapshotManager.finalizeSnapshot(request); - state.reloadStateMachine(lastIncluded); - chunk0CallId.set(-1); + final long i = inProgressInstallSnapshotIndex.getAndUpdate( + previous -> previous == INVALID_LOG_INDEX ? lastIncludedIndex : previous); + if (i != INVALID_LOG_INDEX) { + throw new IOException("Snapshot (index " + i + ") installation is in progress"); + } + try { + stateMachine.pause(); // pause the SM right before publishing the snapshot atomically + snapshotManager.finalizeSnapshot(request); + state.reloadStateMachine(lastIncluded); + chunk0CallId.set(-1); + } finally { + inProgressInstallSnapshotIndex.set(INVALID_LOG_INDEX); + } } } finally { server.updateLastRpcTime(UpdateType.INSTALL_SNAPSHOT_COMPLETE); @@ -295,7 +304,7 @@ private CompletableFuture notifyStateMachineToInstall return future.thenApply(dummy -> reply); } server.getState().getReadRequests().fail( - server.getReadRequestHandler().getReadException("wait for", firstAvailableLogIndex, true)); + server.getReadRequestHandler().getReadException("to wait for readIndex", firstAvailableLogIndex, true)); final RaftPeerProto leaderProto; if (!request.hasLastRaftConfigurationLogEntryProto()) { diff --git a/ratis-server/src/test/java/org/apache/ratis/server/impl/TestStaleReadDuringSnapshotInstallation.java b/ratis-server/src/test/java/org/apache/ratis/server/impl/TestStaleReadDuringSnapshotInstallation.java index c713b9cc24..d2dbac123a 100644 --- a/ratis-server/src/test/java/org/apache/ratis/server/impl/TestStaleReadDuringSnapshotInstallation.java +++ b/ratis-server/src/test/java/org/apache/ratis/server/impl/TestStaleReadDuringSnapshotInstallation.java @@ -22,7 +22,8 @@ import org.apache.ratis.conf.RaftProperties; import org.apache.ratis.protocol.Message; import org.apache.ratis.protocol.RaftClientReply; -import org.apache.ratis.protocol.exceptions.StateMachineException; +import org.apache.ratis.protocol.exceptions.ReadException; +import org.apache.ratis.retry.RetryPolicies; import org.apache.ratis.server.RaftServer; import org.apache.ratis.server.raftlog.RaftLog; import org.apache.ratis.server.simulation.MiniRaftClusterWithSimulatedRpc; @@ -37,10 +38,10 @@ import java.io.IOException; /** - * Test stale reads failing while a snapshot installation is in progress: the follower serving the - * stale read is exactly the node whose state machine is paused/reinitialized by the installation, - * so the read must be rejected with a retriable error instead of querying a paused state machine - * (see {@link ReadRequestHandler#staleReadAsync}). + * Test stale reads being rejected while a snapshot installation is in progress: the follower + * serving the stale read is exactly the node whose state machine is paused/reinitialized by the + * installation, so the read must be rejected with a retriable error instead of querying a paused + * state machine (see {@link ReadRequestHandler#staleReadAsync}). */ public class TestStaleReadDuringSnapshotInstallation { private static final int NUM_SERVERS = 3; @@ -94,13 +95,15 @@ public void testStaleReadRejectedWhileSnapshotInstallationInProgress() throws Ex final RaftClientReply before = client.io().sendStaleRead(message, m4Index, follower.getId()); Assertions.assertTrue(before.isSuccess()); - // Simulate a snapshot installation in progress on the follower. + // Simulate a snapshot installation in progress on the follower: the gate must reject + // the stale read with a retriable ReadException instead of querying the paused state + // machine. Use a no-retry client so the exception surfaces immediately (the default + // test policy would keep retrying it -- see BlockingImpl#sendRequestWithRetry). handler.setInProgressInstallSnapshotIndex(1); - try { + try (final RaftClient noRetryClient = cluster.createClient(follower.getId(), RetryPolicies.noRetry())) { final IOException e = Assertions.assertThrows(IOException.class, - () -> client.io().sendStaleRead(message, m4Index, follower.getId())); - // The gate rejects with a retriable StateMachineException, not a success and not a crash. - Assertions.assertTrue(e instanceof StateMachineException, () -> "Unexpected exception: " + e); + () -> noRetryClient.io().sendStaleRead(message, m4Index, follower.getId())); + Assertions.assertTrue(e instanceof ReadException, () -> "Unexpected exception: " + e); Assertions.assertTrue(e.getMessage().contains("stale read"), () -> e.getMessage()); Assertions.assertTrue(e.getMessage().contains("in progress"), () -> e.getMessage()); } finally {