fix(sync): snapshot maybe change when sending

source/libs/sync/src/syncAppendEntriesReply.c

@@ -165,29 +165,33 @@ int32_t syncNodeOnAppendEntriesReplySnapshotCb(SSyncNode* ths, SyncAppendEntries
 
       // has snapshot
       if (syncNodeHasSnapshot(ths)) {
-        SSnapshot snapshot;
-        ths->pFsm->FpGetSnapshot(ths->pFsm, &snapshot);
-        if (nextIndex <= snapshot.lastApplyIndex) {
-          // ASSERT(nextIndex == snapshot.lastApplyIndex);
+        // get sender
+        SSyncSnapshotSender* pSender = NULL;
+        for (int i = 0; i < ths->replicaNum; ++i) {
+          if (syncUtilSameId(&(pMsg->srcId), &((ths->replicasId)[i]))) {
+            pSender = (ths->senders)[i];
+          }
+        }
+        ASSERT(pSender != NULL);
 
-          nextIndex = snapshot.lastApplyIndex + 1;
-          sInfo("reset new nextIndex %ld, snapshot.lastApplyIndex:%ld", nextIndex, snapshot.lastApplyIndex);
+        // calculate sentryIndex
+        SyncIndex sentryIndex;
+        if (pSender->start) {
+          sentryIndex = pSender->snapshot.lastApplyIndex;
 
+        } else {
           // start send snapshot
-          // get sender
-          SSyncSnapshotSender* pSender = NULL;
-          for (int i = 0; i < ths->replicaNum; ++i) {
-            if (syncUtilSameId(&(pMsg->srcId), &((ths->replicasId)[i]))) {
-              pSender = (ths->senders)[i];
-            }
-          }
-          ASSERT(pSender != NULL);
-
           if (!(pSender->term == ths->pRaftStore->currentTerm && pSender->finish == true)) {
             snapshotSenderStart(pSender);
           } else {
             sInfo("snapshot send finish, send_term:%lu, current_term:%lu", pSender->term, ths->pRaftStore->currentTerm);
           }
+          sentryIndex = pSender->snapshot.lastApplyIndex;
+        }
+
+        // update nextIndex to sentryIndex + 1
+        if (nextIndex <= sentryIndex) {
+          nextIndex = sentryIndex + 1;
         }
       }
 

source/libs/sync/src/syncMain.c

@@ -1693,6 +1693,7 @@ const char* syncStr(ESyncState state) {
 
 int32_t syncNodeCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex, uint64_t flag) {
   int32_t code = 0;
+  sInfo("sync commit from %ld to %ld, flag:0x%lX", beginIndex, endIndex, flag);
 
   // maybe execute by leader, skip snapshot
   SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0};