From 189ae6d801353f8c7583c26d1c7a660c104cf2b6 Mon Sep 17 00:00:00 2001 From: Qian-Cheng-nju Date: Wed, 25 Mar 2026 04:07:47 +0000 Subject: [PATCH 1/3] fix: update commitIndex and lastAppliedIndex after installSnapshot --- .../raft/service/impl/RaftConsensusServiceImpl.java | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/raft-java-core/src/main/java/com/github/wenweihu86/raft/service/impl/RaftConsensusServiceImpl.java b/raft-java-core/src/main/java/com/github/wenweihu86/raft/service/impl/RaftConsensusServiceImpl.java index 4eb69c1..8b59214 100644 --- a/raft-java-core/src/main/java/com/github/wenweihu86/raft/service/impl/RaftConsensusServiceImpl.java +++ b/raft-java-core/src/main/java/com/github/wenweihu86/raft/service/impl/RaftConsensusServiceImpl.java @@ -291,10 +291,17 @@ public RaftProto.InstallSnapshotResponse installSnapshot(RaftProto.InstallSnapsh raftNode.getSnapshot().getLock().unlock(); } - // discard old log entries + // discard old log entries and update node state raftNode.getLock().lock(); try { raftNode.getRaftLog().truncatePrefix(lastSnapshotIndex + 1); + raftNode.setCommitIndex(lastSnapshotIndex); + raftNode.setLastAppliedIndex(lastSnapshotIndex); + RaftProto.Configuration snapshotConfiguration + = raftNode.getSnapshot().getMetaData().getConfiguration(); + if (snapshotConfiguration.getServersCount() > 0) { + raftNode.setConfiguration(snapshotConfiguration); + } } finally { raftNode.getLock().unlock(); } From 28db4a22e60621491498d11580d82e89e63db55a Mon Sep 17 00:00:00 2001 From: Qian-Cheng-nju Date: Wed, 25 Mar 2026 04:19:34 +0000 Subject: [PATCH 2/3] test: add reproduction test for installSnapshot state update bug --- .../service/impl/InstallSnapshotBugTest.java | 211 ++++++++++++++++++ 1 file changed, 211 insertions(+) create mode 100644 raft-java-core/src/test/java/com/github/wenweihu86/raft/service/impl/InstallSnapshotBugTest.java diff --git a/raft-java-core/src/test/java/com/github/wenweihu86/raft/service/impl/InstallSnapshotBugTest.java b/raft-java-core/src/test/java/com/github/wenweihu86/raft/service/impl/InstallSnapshotBugTest.java new file mode 100644 index 0000000..df0ca23 --- /dev/null +++ b/raft-java-core/src/test/java/com/github/wenweihu86/raft/service/impl/InstallSnapshotBugTest.java @@ -0,0 +1,211 @@ +package com.github.wenweihu86.raft.service.impl; + +import com.github.wenweihu86.raft.RaftNode; +import com.github.wenweihu86.raft.RaftOptions; +import com.github.wenweihu86.raft.StateMachine; +import com.github.wenweihu86.raft.proto.RaftProto; +import com.google.protobuf.ByteString; +import org.apache.commons.io.FileUtils; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import java.io.File; +import java.io.IOException; +import java.lang.reflect.Field; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; + +/** + * Reproduction test for raft-java bug #75: + * + * After installSnapshot completes (final chunk), the handler applies the + * state machine from the snapshot and truncates the log, but FAILS to + * update commitIndex, lastAppliedIndex, or configuration to match the + * snapshot's lastIncludedIndex and included configuration. + * + * Compare with the constructor (RaftNode lines 88-112) which correctly + * sets commitIndex = max(snapshot.lastIncludedIndex, log.commitIndex), + * lastAppliedIndex = commitIndex, and configuration = snapshot config. + * + * The buggy code is in RaftConsensusServiceImpl.installSnapshot(), + * lines 279-301, where after the final chunk: + * - stateMachine.readSnapshot() is called (OK) + * - snapshot.reload() is called (OK) + * - raftLog.truncatePrefix() is called (OK) + * - commitIndex is NOT updated (BUG) + * - lastAppliedIndex is NOT updated (BUG) + * - configuration is NOT updated (BUG) + */ +public class InstallSnapshotBugTest { + + private static final String TEST_DATA_DIR = "/tmp/raft-java-bug75-test"; + + private RaftNode raftNode; + private RaftConsensusServiceImpl service; + private NoOpStateMachine stateMachine; + private ScheduledExecutorService scheduler; + + /** Minimal state machine that does nothing -- sufficient for this test. */ + static class NoOpStateMachine implements StateMachine { + boolean readSnapshotCalled = false; + @Override public void writeSnapshot(String snapshotDir) {} + @Override public void readSnapshot(String snapshotDir) { + readSnapshotCalled = true; + } + @Override public void apply(byte[] dataBytes) {} + } + + @Before + public void setUp() throws Exception { + // Clean up any previous test data + FileUtils.deleteDirectory(new File(TEST_DATA_DIR)); + + // Build initial configuration: 3 servers (server IDs 1, 2, 3) + List initialServers = new ArrayList<>(); + for (int i = 1; i <= 3; i++) { + initialServers.add(RaftProto.Server.newBuilder() + .setServerId(i) + .setEndpoint(RaftProto.Endpoint.newBuilder() + .setHost("127.0.0.1") + .setPort(8000 + i) + .build()) + .build()); + } + + RaftOptions options = new RaftOptions(); + options.setDataDir(TEST_DATA_DIR); + // Use large election timeout to avoid interference + options.setElectionTimeoutMilliseconds(60000); + options.setSnapshotPeriodSeconds(3600); + + stateMachine = new NoOpStateMachine(); + + // localServer is server 1 + raftNode = new RaftNode(options, initialServers, initialServers.get(0), stateMachine); + + // Instead of calling init() (which tries to create BRPC peer connections), + // inject a ScheduledExecutorService via reflection so that stepDown() -> + // resetElectionTimer() does not NPE. + scheduler = Executors.newScheduledThreadPool(2); + Field schedulerField = RaftNode.class.getDeclaredField("scheduledExecutorService"); + schedulerField.setAccessible(true); + schedulerField.set(raftNode, scheduler); + + // Simulate some prior progress: commitIndex=5, lastAppliedIndex=5 + raftNode.setCommitIndex(5); + raftNode.setLastAppliedIndex(5); + + service = new RaftConsensusServiceImpl(raftNode); + } + + @After + public void tearDown() throws Exception { + if (scheduler != null) { + scheduler.shutdownNow(); + } + FileUtils.deleteDirectory(new File(TEST_DATA_DIR)); + } + + @Test + public void testInstallSnapshotDoesNotUpdateCommitIndexOrConfiguration() throws IOException { + // --- Arrange --- + + // Build a snapshot configuration with 4 servers (different from initial 3) + RaftProto.Configuration.Builder snapshotConfBuilder = RaftProto.Configuration.newBuilder(); + for (int i = 1; i <= 4; i++) { + snapshotConfBuilder.addServers(RaftProto.Server.newBuilder() + .setServerId(i) + .setEndpoint(RaftProto.Endpoint.newBuilder() + .setHost("127.0.0.1") + .setPort(9000 + i) + .build()) + .build()); + } + RaftProto.Configuration snapshotConf = snapshotConfBuilder.build(); + + // Build snapshot metadata: lastIncludedIndex=100, lastIncludedTerm=10 + RaftProto.SnapshotMetaData snapshotMeta = RaftProto.SnapshotMetaData.newBuilder() + .setLastIncludedIndex(100) + .setLastIncludedTerm(10) + .setConfiguration(snapshotConf) + .build(); + + // The installSnapshot request: single chunk (isFirst=true, isLast=true) + // term must be >= currentTerm to pass the term check + RaftProto.InstallSnapshotRequest request = RaftProto.InstallSnapshotRequest.newBuilder() + .setServerId(2) // leader is server 2 + .setTerm(10) // high enough term + .setSnapshotMetaData(snapshotMeta) + .setFileName("state.dat") + .setOffset(0) + .setData(ByteString.copyFromUtf8("snapshot-data")) + .setIsFirst(true) + .setIsLast(true) + .build(); + + // Record pre-install state + long preCommitIndex = raftNode.getCommitIndex(); + long preLastApplied = raftNode.getLastAppliedIndex(); + int preConfigSize = raftNode.getConfiguration().getServersCount(); + + Assert.assertEquals("Pre-condition: commitIndex should be 5", 5, preCommitIndex); + Assert.assertEquals("Pre-condition: lastAppliedIndex should be 5", 5, preLastApplied); + Assert.assertEquals("Pre-condition: configuration should have 3 servers", 3, preConfigSize); + + // --- Act --- + RaftProto.InstallSnapshotResponse response = service.installSnapshot(request); + + // --- Assert --- + Assert.assertEquals("installSnapshot should succeed", + RaftProto.ResCode.RES_CODE_SUCCESS, response.getResCode()); + + // Verify the state machine WAS applied (this part works correctly) + Assert.assertTrue("State machine readSnapshot should have been called", + stateMachine.readSnapshotCalled); + + // Verify the snapshot metadata was reloaded (this part works correctly) + long snapshotLastIndex = raftNode.getSnapshot().getMetaData().getLastIncludedIndex(); + Assert.assertEquals("Snapshot metadata should reflect lastIncludedIndex=100", + 100, snapshotLastIndex); + + // === BUG DEMONSTRATION === + // After installSnapshot, commitIndex SHOULD be updated to at least + // lastIncludedIndex (100), just as the constructor does on startup. + // But the buggy code leaves it at the old value. + long postCommitIndex = raftNode.getCommitIndex(); + long postLastApplied = raftNode.getLastAppliedIndex(); + int postConfigSize = raftNode.getConfiguration().getServersCount(); + + // These assertions demonstrate the bug: values are STILL at old levels + System.out.println("=== Bug #75 Demonstration ==="); + System.out.println("Snapshot lastIncludedIndex: 100"); + System.out.println("commitIndex after install: " + postCommitIndex + " (expected: 100)"); + System.out.println("lastAppliedIndex after install: " + postLastApplied + " (expected: 100)"); + System.out.println("configuration servers after install: " + postConfigSize + " (expected: 4)"); + + // BUG: commitIndex is still 5 instead of 100 + Assert.assertEquals( + "BUG: commitIndex was not updated after installSnapshot " + + "(should be 100, still " + postCommitIndex + ")", + 5, postCommitIndex); + + // BUG: lastAppliedIndex is still 5 instead of 100 + Assert.assertEquals( + "BUG: lastAppliedIndex was not updated after installSnapshot " + + "(should be 100, still " + postLastApplied + ")", + 5, postLastApplied); + + // BUG: configuration still has 3 servers instead of 4 + Assert.assertEquals( + "BUG: configuration was not updated after installSnapshot " + + "(should be 4 servers, still " + postConfigSize + ")", + 3, postConfigSize); + + System.out.println("\nAll assertions passed: bug #75 is confirmed."); + System.out.println("commitIndex, lastAppliedIndex, and configuration are NOT updated."); + } +} From 82ed69ef071ce1ef9adee987b598c2053354b40f Mon Sep 17 00:00:00 2001 From: Qian-Cheng-nju Date: Wed, 25 Mar 2026 04:45:20 +0000 Subject: [PATCH 3/3] test: fix assertions to verify fix instead of demonstrating bug --- .../service/impl/InstallSnapshotBugTest.java | 41 ++++++++----------- 1 file changed, 18 insertions(+), 23 deletions(-) diff --git a/raft-java-core/src/test/java/com/github/wenweihu86/raft/service/impl/InstallSnapshotBugTest.java b/raft-java-core/src/test/java/com/github/wenweihu86/raft/service/impl/InstallSnapshotBugTest.java index df0ca23..94f8f82 100644 --- a/raft-java-core/src/test/java/com/github/wenweihu86/raft/service/impl/InstallSnapshotBugTest.java +++ b/raft-java-core/src/test/java/com/github/wenweihu86/raft/service/impl/InstallSnapshotBugTest.java @@ -172,40 +172,35 @@ public void testInstallSnapshotDoesNotUpdateCommitIndexOrConfiguration() throws Assert.assertEquals("Snapshot metadata should reflect lastIncludedIndex=100", 100, snapshotLastIndex); - // === BUG DEMONSTRATION === - // After installSnapshot, commitIndex SHOULD be updated to at least - // lastIncludedIndex (100), just as the constructor does on startup. - // But the buggy code leaves it at the old value. + // === FIX VERIFICATION === + // After the fix, commitIndex, lastAppliedIndex, and configuration + // should all be updated to match the snapshot, just as the constructor + // does on startup (RaftNode.java lines 88-112). long postCommitIndex = raftNode.getCommitIndex(); long postLastApplied = raftNode.getLastAppliedIndex(); int postConfigSize = raftNode.getConfiguration().getServersCount(); - // These assertions demonstrate the bug: values are STILL at old levels - System.out.println("=== Bug #75 Demonstration ==="); + System.out.println("=== Bug #75 Fix Verification ==="); System.out.println("Snapshot lastIncludedIndex: 100"); - System.out.println("commitIndex after install: " + postCommitIndex + " (expected: 100)"); - System.out.println("lastAppliedIndex after install: " + postLastApplied + " (expected: 100)"); - System.out.println("configuration servers after install: " + postConfigSize + " (expected: 4)"); + System.out.println("commitIndex after install: " + postCommitIndex); + System.out.println("lastAppliedIndex after install: " + postLastApplied); + System.out.println("configuration servers after install: " + postConfigSize); - // BUG: commitIndex is still 5 instead of 100 + // After fix: commitIndex should be updated to 100 Assert.assertEquals( - "BUG: commitIndex was not updated after installSnapshot " - + "(should be 100, still " + postCommitIndex + ")", - 5, postCommitIndex); + "commitIndex should be updated to snapshot lastIncludedIndex", + 100, postCommitIndex); - // BUG: lastAppliedIndex is still 5 instead of 100 + // After fix: lastAppliedIndex should be updated to 100 Assert.assertEquals( - "BUG: lastAppliedIndex was not updated after installSnapshot " - + "(should be 100, still " + postLastApplied + ")", - 5, postLastApplied); + "lastAppliedIndex should be updated to snapshot lastIncludedIndex", + 100, postLastApplied); - // BUG: configuration still has 3 servers instead of 4 + // After fix: configuration should have 4 servers from snapshot Assert.assertEquals( - "BUG: configuration was not updated after installSnapshot " - + "(should be 4 servers, still " + postConfigSize + ")", - 3, postConfigSize); + "configuration should be updated from snapshot metadata", + 4, postConfigSize); - System.out.println("\nAll assertions passed: bug #75 is confirmed."); - System.out.println("commitIndex, lastAppliedIndex, and configuration are NOT updated."); + System.out.println("\nAll assertions passed: fix verified."); } }