fix(store): resume learner promotion on retry

hugegraph-store/docs/operations-guide.md

@@ -654,6 +654,17 @@ curl http://192.168.1.10:8620/v1/partitionsAndStatus
 
 5. **Verify**: Wait for even distribution (may take hours)
 
+Replica migration can leave learners in a Raft group while they receive a snapshot or catch up.
+If the same membership change is retried after a catch-up timeout, Store resumes those learners
+and promotes them to voting members after catch-up; it does not need to create them again.
+Each catch-up check waits up to 600 seconds before returning `TASK_CONTINUE`. This is a
+per-attempt wait, not an overall migration deadline; a peer that cannot catch up can keep
+the operation pending across retries. Inspect its replication progress and errors before
+continuing a node operation. This retry correction does not add durable task admission
+or recovery of a migration task across a Leader change.
+Verify the actual voter and learner lists before treating migration as complete: a target count
+that includes learners is not proof of voting membership.
+
 ### Removing Store Nodes
 
 **When to Remove**:

hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/PartitionEngine.java

@@ -295,23 +295,26 @@ public ShardGroup getShardGroup() {
      * @return true means completed, false means not completed
      */
     public Status changePeers(List<String> peers, final Closure done) {
-        if (ListUtils.isEqualList(peers, RaftUtils.getPeerEndpoints(raftNode))) {
+        Configuration oldConf = getCurrentConf();
+        List<String> oldVoters = RaftUtils.getPeerEndpoints(oldConf);
+        if (ListUtils.isEqualList(peers, oldVoters) && oldConf.listLearners().isEmpty()) {
             return Status.OK();
         }
 
         Status result = HgRaftError.TASK_CONTINUE.toStatus();
-        List<String> oldPeers = RaftUtils.getAllEndpoints(raftNode);
+        List<String> oldPeers = RaftUtils.getAllEndpoints(oldConf);
         log.info("Raft {} changePeers start, old peer is {}, new peer is {}",
                  getGroupId(), oldPeers, peers);
         // Check the peer that needs to be added.
         List<String> addPeers = ListUtils.removeAll(peers, oldPeers);
+        // A retry must still promote target peers that already exist as learners.
+        List<String> pendingPromotions = ListUtils.removeAll(peers, oldVoters);
         // learner to be deleted. Possible peer change.
         List<String> removedPeers = ListUtils.removeAll(oldPeers, peers);
 
         HgCmdClient rpcClient = storeEngine.getHgCmdClient();
         // Generate a new Configuration object
 
-        Configuration oldConf = getCurrentConf();
         Configuration conf = oldConf.copy();
 
         FutureClosure closure;
@@ -341,31 +344,37 @@ public Status changePeers(List<String> peers, final Closure done) {
                          status.getErrorMsg());
                 return status;
             }
+        }
 
-            // 2.2 Waiting learner to synchronize snapshot (check added learner)
-            if (!waitForReplicate(addPeers)) {
+        if (!pendingPromotions.isEmpty()) {
+            // 2.2 Wait for both newly added and previously created target learners.
+            if (!waitForReplicate(pendingPromotions)) {
                 return HgRaftError.TASK_CONTINUE.toStatus();
             }
 
             log.info("Raft {} replicate status is OK", getGroupId());
 
             closure = new FutureClosure();
             // 2.3 change learner to follower (first remove, then add follower)
-            raftNode.removeLearners(conf.listLearners(), closure);
-            if (!closure.get().isOk()) {
+            List<PeerId> promoting = new ArrayList<>();
+            pendingPromotions.forEach(peer -> promoting.add(JRaftUtils.getPeerId(peer)));
+            raftNode.removeLearners(promoting, closure);
+            var status = closure.get();
+            if (!status.isOk()) {
                 log.error("Raft {} remove learner error, result:{}", getGroupId(), status);
                 return HgRaftError.TASK_ERROR.toStatus();
             }
 
-            addPeers.forEach(peer -> {
+            pendingPromotions.forEach(peer -> {
                 conf.removeLearner(JRaftUtils.getPeerId(peer));
                 conf.addPeer(JRaftUtils.getPeerId(peer));
             });
 
             // add follower
             closure = new FutureClosure();
             raftNode.changePeers(conf, closure);
-            if (!closure.get().isOk()) {
+            status = closure.get();
+            if (!status.isOk()) {
                 log.error("Raft {} changePeers error, result:{}", getGroupId(), status);
                 return HgRaftError.TASK_ERROR.toStatus();
             }
@@ -419,7 +428,7 @@ public Status changePeers(List<String> peers, final Closure done) {
      *
      * @return false if the peers did not catch up within 600s
      */
-    private boolean waitForReplicate(List<String> peers) {
+    boolean waitForReplicate(List<String> peers) {
         //todo Each learner will wait for 1s, if another one is not sync.Consider using
         // countdownLatch
         boolean allLearnerSnapshotOk = false;

hugegraph-store/hg-store-test/pom.xml

@@ -246,6 +246,7 @@
                                 <include>**/IpUtilTest.java</include>
                                 <include>**/PartitionStateMachineTest.java</include>
                                 <include>**/PartitionEngineErrorTest.java</include>
+                                <include>**/PartitionEngineMembershipTest.java</include>
                                 <include>**/BatchGraphIsolationTest.java</include>
                                 <include>**/GraphIDManagerTest.java</include>
                                 <include>**/SchemaDriverTest.java</include>