fix: lease read linearizability in RaftUtil.incrTerm()

dtprj · Jun 18, 2024 · 5f22757 · 5f22757
1 parent 6388064
commit 5f22757
Showing 1 changed file with 11 additions and 4 deletions.
diff --git a/server/src/main/java/com/github/dtprj/dongting/raft/impl/RaftUtil.java b/server/src/main/java/com/github/dtprj/dongting/raft/impl/RaftUtil.java
@@ -15,6 +15,7 @@
  */
 package com.github.dtprj.dongting.raft.impl;
 
+import com.github.dtprj.dongting.common.Pair;
 import com.github.dtprj.dongting.common.RefCount;
 import com.github.dtprj.dongting.fiber.Fiber;
 import com.github.dtprj.dongting.fiber.FiberGroup;
@@ -34,6 +35,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashSet;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -123,21 +125,22 @@ public static void incrTerm(int remoteTerm, RaftStatusImpl raftStatus, int newLe
         if (newLeaderId > 0) {
             updateLeader(raftStatus, newLeaderId);
         }
+        LinkedList<Pair<CompletableFuture<?>, NotLeaderException>> failList = new LinkedList<>();
         if (oldRole != RaftRole.observer) {
             log.info("update term from {} to {}, change to follower, oldRole={}",
                     raftStatus.getCurrentTerm(), remoteTerm, raftStatus.getRole());
-            TailCache oldPending = raftStatus.getTailCache();
             raftStatus.setRole(RaftRole.follower);
             if (oldRole == RaftRole.leader) {
+                TailCache oldPending = raftStatus.getTailCache();
+                NotLeaderException e = new NotLeaderException(raftStatus.getCurrentLeaderNode());
                 oldPending.forEach((idx, task) -> {
-                    RaftNode leaderNode = raftStatus.getCurrentLeaderNode();
                     if (task.getFuture() != null) {
-                        task.getFuture().completeExceptionally(new NotLeaderException(leaderNode));
+                        failList.add(new Pair<>(task.getFuture(), e));
                     }
                     RaftTask reader;
                     while ((reader = task.getNextReader()) != null) {
                         if (reader.getFuture() != null) {
-                            reader.getFuture().completeExceptionally(new NotLeaderException(leaderNode));
+                            failList.add(new Pair<>(reader.getFuture(), e));
                         }
                     }
                 });
@@ -149,6 +152,10 @@ public static void incrTerm(int remoteTerm, RaftStatusImpl raftStatus, int newLe
         raftStatus.setCurrentTerm(remoteTerm);
         raftStatus.setVotedFor(0);
         raftStatus.copyShareStatus();
+        // copy share status should happen before futures complete
+        for(Pair<CompletableFuture<?>, NotLeaderException> pair : failList) {
+            pair.getLeft().completeExceptionally(pair.getRight());
+        }
     }
 
     public static void resetStatus(RaftStatusImpl raftStatus) {