fix: catch uncaught exception in onApply to prevent apply pipeline stall (#1246)

Qian-Cheng-nju · web-flow · commit 913a85ab624f · 2026-04-08T16:43:40.000+08:00
* fix: catch uncaught exception in onApply to prevent apply pipeline stall

* use setError instead of setErrorAndRollback per maintainer feedback

* fix: rewrite flaky tests to avoid Thread.sleep and leader shutdown races

---------

Co-authored-by: Qian-Cheng-nju &lt;Qian-Cheng-nju@users.noreply.github.com&gt;
diff --git a/jraft-core/src/main/java/com/alipay/sofa/jraft/core/FSMCallerImpl.java b/jraft-core/src/main/java/com/alipay/sofa/jraft/core/FSMCallerImpl.java
@@ -559,7 +559,19 @@ private void doCommitted(final long committedIndex) {
                 }
 
                 // Apply data task to user state machine
-                doApplyTasks(iterImpl);
+                try {
+                    doApplyTasks(iterImpl);
+                } catch (final Throwable t) {
+                    // If the user state machine throws, we must not let the
+                    // exception propagate — that would skip setLastApplied()
+                    // and stall the apply pipeline permanently (the same
+                    // entries would be retried on every doCommitted() call,
+                    // hitting the same exception in an infinite loop).
+                    LOG.error("StateMachine threw when applying entries starting at index={}. "
+                              + "Halting state machine to avoid stall.", iterImpl.getIndex(), t);
+                    iterImpl.setError(new Status(RaftError.ESTATEMACHINE, "StateMachine threw: %s", t.getMessage()));
+                    break;
+                }
             }
 
             if (iterImpl.hasError()) {
diff --git a/jraft-core/src/main/java/com/alipay/sofa/jraft/core/IteratorImpl.java b/jraft-core/src/main/java/com/alipay/sofa/jraft/core/IteratorImpl.java
@@ -167,6 +167,19 @@ public void commitAndSnapshotSync(Closure done) {
         }
     }
 
+    /**
+     * Set error without rolling back the index. Used when the state machine
+     * throws an uncaught exception — we want to halt the node (ERROR state)
+     * without assuming any rollback policy.
+     */
+    public void setError(final Status st) {
+        this.currEntry = null;
+        getOrCreateError().setType(EnumOutter.ErrorType.ERROR_TYPE_STATE_MACHINE);
+        getOrCreateError().getStatus().setError(RaftError.ESTATEMACHINE,
+            "StateMachine meet critical error when applying one or more tasks since index=%d, %s", this.currentIndex,
+            st != null ? st.toString() : "none");
+    }
+
     public void setErrorAndRollback(final long ntail, final Status st) {
         Requires.requireTrue(ntail > 0, "Invalid ntail=" + ntail);
         if (this.currEntry == null || this.currEntry.getType() != EnumOutter.EntryType.ENTRY_TYPE_DATA) {
diff --git a/jraft-core/src/test/java/com/alipay/sofa/jraft/core/ElectSelfPersistOrderTest.java b/jraft-core/src/test/java/com/alipay/sofa/jraft/core/ElectSelfPersistOrderTest.java
@@ -17,7 +17,6 @@
 package com.alipay.sofa.jraft.core;
 
 import java.io.File;
-import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
@@ -33,10 +32,8 @@
 import com.alipay.sofa.jraft.Iterator;
 import com.alipay.sofa.jraft.Node;
 import com.alipay.sofa.jraft.RaftGroupService;
-import com.alipay.sofa.jraft.Status;
 import com.alipay.sofa.jraft.conf.Configuration;
 import com.alipay.sofa.jraft.entity.PeerId;
-import com.alipay.sofa.jraft.entity.Task;
 import com.alipay.sofa.jraft.option.NodeOptions;
 import com.alipay.sofa.jraft.option.RaftMetaStorageOptions;
 import com.alipay.sofa.jraft.option.RaftOptions;
@@ -86,16 +83,17 @@ public void teardown() throws Exception {
 
     /**
      * Verify that when setTermAndVotedFor() fails during electSelf(), the node
-     * steps down and does NOT send RequestVote RPCs to peers.
+     * steps down and does NOT become leader.
      *
      * Approach:
      * 1. Start a 3-node cluster with a FailingMetaStorage on node 0
-     * 2. Wait for leader election to complete normally
+     * 2. Wait for leader election to complete normally (node 0 must not be leader)
      * 3. Enable the failure flag on node 0's meta storage
-     * 4. Stop the current leader to force new elections
-     * 5. Node 0 will attempt electSelf(), fail on setTermAndVotedFor(), step down
-     * 6. Verify the observer node's term did NOT advance due to node 0's failed election
-     *    (proving no RequestVote RPC was sent before the persistence check)
+     * 4. Directly trigger electSelf() on node 0 via tryElectSelf()
+     * 5. Verify: persist failure detected, node 0 did NOT become leader
+     *
+     * Using tryElectSelf() directly avoids the need to kill the leader (whose
+     * graceful shutdown sends TimeoutNow, independently bumping follower terms).
      */
     @Test
     public void testPersistFailurePreventsRpcSend() throws Exception {
@@ -113,9 +111,6 @@ public void testPersistFailurePreventsRpcSend() throws Exception {
         for (int i = 0; i < peers.size(); i++) {
             final PeerId peer = peers.get(i);
             final NodeOptions nodeOptions = new NodeOptions();
-            // node0 gets a short election timeout so it fires first after leader death.
-            // Other nodes get a long timeout so they won't independently start elections
-            // during the test window — this lets us assert term equality on the observer.
             nodeOptions.setElectionTimeoutMs(i == 0 ? 300 : 10_000);
             nodeOptions.setSnapshotIntervalSecs(300);
             nodeOptions.setInitialConf(conf);
@@ -149,91 +144,37 @@ public void onApply(final Iterator iter) {
 
         try {
             // Wait for initial leader election (fail flag is off)
-            Thread.sleep(3000);
-
-            Node leader = null;
-            for (final Node n : nodes) {
-                if (n.getLeaderId() != null && !n.getLeaderId().isEmpty()) {
-                    if (n.getNodeId().getPeerId().equals(n.getLeaderId())) {
-                        leader = n;
-                    }
-                }
-            }
+            Node leader = waitForLeader(nodes, null, 30_000);
             assertNotNull("Should have a leader", leader);
 
             final Node node0 = nodes.get(0);
 
             // If node0 is the leader, transfer leadership away
             if (leader == node0) {
                 leader.transferLeadershipTo(peers.get(1));
-                Thread.sleep(2000);
-                leader = null;
-                for (final Node n : nodes) {
-                    if (n.getLeaderId() != null && !n.getLeaderId().isEmpty()) {
-                        if (n.getNodeId().getPeerId().equals(n.getLeaderId())) {
-                            leader = n;
-                        }
-                    }
-                }
-                assumeTrue("Could not transfer leadership away from node0", leader != node0);
-            }
-
-            assertNotNull("Should still have a leader", leader);
-            final PeerId leaderPeer = leader.getLeaderId();
-            final int leaderIdx = peers.indexOf(leaderPeer);
-
-            // Find the observer (not node0, not leader)
-            Node observer = null;
-            int observerIdx = -1;
-            for (int i = 0; i < nodes.size(); i++) {
-                if (i != 0 && i != leaderIdx) {
-                    observer = nodes.get(i);
-                    observerIdx = i;
-                    break;
-                }
+                leader = waitForLeader(nodes, node0, 30_000);
+                assumeTrue("Could not transfer leadership away from node0", leader != null && leader != node0);
             }
-            assertNotNull("Should have an observer", observer);
-
-            final long observerTermBefore = ((NodeImpl) observer).getCurrentTerm();
 
             // Enable fail flag: next setTermAndVotedFor on node0 returns false
             failFlag.set(true);
 
-            // Stop leader to trigger elections
-            services.get(leaderIdx).shutdown();
-            services.get(leaderIdx).join();
-
-            // Wait for node0 to attempt election and hit the persist failure
-            boolean failed = failLatch.await(10, TimeUnit.SECONDS);
-            assertTrue("Node0 should have attempted election", failed);
-
-            final long failTerm = failedAtTerm.get();
-
-            // Brief pause for any in-flight messages
-            Thread.sleep(500);
-
-            final long observerTermAfter = ((NodeImpl) observer).getCurrentTerm();
-
-            // With the fix: setTermAndVotedFor is called BEFORE the RPC loop.
-            // When it returns false, electSelf() steps down and returns without
-            // sending any RequestVote RPCs. Therefore the observer's term should
-            // NOT have been bumped by node0's failed election attempt.
-            //
-            // Without the fix: RPCs were sent BEFORE setTermAndVotedFor, so the
-            // observer would have already received the RequestVote and bumped its
-            // term even though node0's persistence failed.
-            //
-            // The observer uses a long election timeout (10s) so it won't
-            // independently start its own election during this short test window.
-            System.out.println("Observer term before=" + observerTermBefore + ", after=" + observerTermAfter
-                               + ", node0 failed at term=" + failTerm);
-
-            // Observer's term must not change — proves no RequestVote RPC was sent.
-            assertEquals("Observer term should not advance (no RequestVote RPC sent)", observerTermBefore,
-                observerTermAfter);
-
-            // With the fix, node0's persist failure means it steps down and
-            // does NOT send RequestVote RPCs. Verify node0 is no longer leader.
+            // Directly trigger electSelf() on node0. With the fix, electSelf()
+            // calls setTermAndVotedFor() BEFORE sending RequestVote RPCs. When
+            // it returns false, electSelf() steps down without sending any RPCs.
+            // Without the fix, RPCs would be sent first and node0 could become
+            // leader despite the persistence failure (since quorum = 2 and the
+            // observer would grant the vote).
+            ((NodeImpl) node0).tryElectSelf();
+
+            // Persist failure should have been detected
+            assertTrue("Node0 should have hit persist failure", failLatch.await(5, TimeUnit.SECONDS));
+            assertTrue("Failed term should be positive", failedAtTerm.get() > 0);
+
+            System.out.println("Node0 persist failed at term=" + failedAtTerm.get());
+
+            // Node0 must NOT be leader — proves electSelf() did not proceed
+            // past the failed setTermAndVotedFor() to send RPCs and win votes.
             assertFalse("Node0 should not be leader after persist failure", ((NodeImpl) node0).isLeader());
 
         } finally {
@@ -247,6 +188,20 @@ public void onApply(final Iterator iter) {
         }
     }
 
+    private static Node waitForLeader(List<Node> nodes, Node excluded, long timeoutMs) throws InterruptedException {
+        final long deadline = System.currentTimeMillis() + timeoutMs;
+        while (System.currentTimeMillis() < deadline) {
+            for (final Node n : nodes) {
+                if (n != excluded && n.getLeaderId() != null && !n.getLeaderId().isEmpty()
+                    && n.getNodeId().getPeerId().equals(n.getLeaderId())) {
+                    return n;
+                }
+            }
+            Thread.sleep(100);
+        }
+        return null;
+    }
+
     /**
      * A RaftMetaStorage wrapper that returns false from setTermAndVotedFor()
      * when the fail flag is set, simulating a disk I/O error during election.
diff --git a/jraft-core/src/test/java/com/alipay/sofa/jraft/core/OnApplyExceptionTest.java b/jraft-core/src/test/java/com/alipay/sofa/jraft/core/OnApplyExceptionTest.java
