Merge pull request #14051 from taosdata/feature/3.0_mhli

fix(sync): config change when set index-mgr

source/libs/sync/inc/syncInt.h

@@ -253,6 +253,7 @@ void syncNodePrint(SSyncNode* pObj);
 void syncNodePrint2(char* s, SSyncNode* pObj);
 void syncNodeLog(SSyncNode* pObj);
 void syncNodeLog2(char* s, SSyncNode* pObj);
+void syncNodeLog3(char* s, SSyncNode* pObj);
 
 #ifdef __cplusplus
 }

source/libs/sync/inc/syncRaftCfg.h

@@ -27,7 +27,7 @@ extern "C" {
 #include "syncInt.h"
 #include "taosdef.h"
 
-#define CONFIG_FILE_LEN 1024
+#define CONFIG_FILE_LEN 2048
 
 #define MAX_CONFIG_INDEX_COUNT 512
 

source/libs/sync/inc/syncSnapshot.h

@@ -55,7 +55,7 @@ typedef struct SSyncSnapshotSender {
 SSyncSnapshotSender *snapshotSenderCreate(SSyncNode *pSyncNode, int32_t replicaIndex);
 void                 snapshotSenderDestroy(SSyncSnapshotSender *pSender);
 bool                 snapshotSenderIsStart(SSyncSnapshotSender *pSender);
-void                 snapshotSenderStart(SSyncSnapshotSender *pSender);
+void                 snapshotSenderStart(SSyncSnapshotSender *pSender, SSnapshot snapshot, void *pReader);
 void                 snapshotSenderStop(SSyncSnapshotSender *pSender);
 int32_t              snapshotSend(SSyncSnapshotSender *pSender);
 int32_t              snapshotReSend(SSyncSnapshotSender *pSender);

source/libs/sync/src/syncAppendEntriesReply.c

@@ -173,6 +173,28 @@ int32_t syncNodeOnAppendEntriesReplySnapshotCb(SSyncNode* ths, SyncAppendEntries
       // get sender
       SSyncSnapshotSender* pSender = syncNodeGetSnapshotSender(ths, &(pMsg->srcId));
       ASSERT(pSender != NULL);
+
+      SSnapshot snapshot;
+      void*     pReader = NULL;
+      ths->pFsm->FpGetSnapshot(ths->pFsm, &snapshot, NULL, &pReader);
+      if (snapshot.lastApplyIndex >= SYNC_INDEX_BEGIN && nextIndex <= snapshot.lastApplyIndex + 1 &&
+          !snapshotSenderIsStart(pSender) && pMsg->privateTerm < pSender->privateTerm) {
+        // has snapshot
+        ASSERT(pReader != NULL);
+        snapshotSenderStart(pSender, snapshot, pReader);
+
+        char* eventLog = snapshotSender2SimpleStr(pSender, "snapshot sender start");
+        syncNodeEventLog(ths, eventLog);
+        taosMemoryFree(eventLog);
+
+      } else {
+        // no snapshot
+        if (pReader != NULL) {
+          ths->pFsm->FpSnapshotStopRead(ths->pFsm, pReader);
+        }
+      }
+
+      /*
             bool      hasSnapshot = syncNodeHasSnapshot(ths);
             SSnapshot snapshot;
             ths->pFsm->FpGetSnapshotInfo(ths->pFsm, &snapshot);
@@ -187,6 +209,7 @@ int32_t syncNodeOnAppendEntriesReplySnapshotCb(SSyncNode* ths, SyncAppendEntries
               syncNodeEventLog(ths, eventLog);
               taosMemoryFree(eventLog);
             }
+      */
 
       SyncIndex sentryIndex = pSender->snapshot.lastApplyIndex + 1;
 
@@ -207,12 +230,6 @@ int32_t syncNodeOnAppendEntriesReplySnapshotCb(SSyncNode* ths, SyncAppendEntries
 
   syncIndexMgrLog2("recv SyncAppendEntriesReply, after pNextIndex:", ths->pNextIndex);
   syncIndexMgrLog2("recv SyncAppendEntriesReply, after pMatchIndex:", ths->pMatchIndex);
-  if (gRaftDetailLog) {
-    SSnapshot snapshot;
-    ths->pFsm->FpGetSnapshotInfo(ths->pFsm, &snapshot);
-    sTrace("recv SyncAppendEntriesReply, after snapshot.lastApplyIndex:%ld, snapshot.lastApplyTerm:%lu",
-           snapshot.lastApplyIndex, snapshot.lastApplyTerm);
-  }
 
   return ret;
 }

source/libs/sync/src/syncIndexMgr.c

@@ -63,7 +63,12 @@ void syncIndexMgrSetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId,
   }
 
   // maybe config change
-  assert(0);
+  // assert(0);
+
+  char     host[128];
+  uint16_t port;
+  syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port);
+  sError("vgId:%d index mgr set for %s:%d, index:%" PRId64 " error", pSyncIndexMgr->pSyncNode->vgId, host, port, index);
 }
 
 SyncIndex syncIndexMgrGetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId) {
@@ -73,7 +78,9 @@ SyncIndex syncIndexMgrGetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaf
       return idx;
     }
   }
-  assert(0);
+
+  syncNodeLog3("syncIndexMgrGetIndex", pSyncIndexMgr->pSyncNode);
+  ASSERT(0);
 }
 
 cJSON *syncIndexMgr2Json(SSyncIndexMgr *pSyncIndexMgr) {
@@ -162,7 +169,11 @@ void syncIndexMgrSetTerm(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, S
   }
 
   // maybe config change
-  assert(0);
+  // assert(0);
+  char     host[128];
+  uint16_t port;
+  syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port);
+  sError("vgId:%d index mgr set for %s:%d, term:%lu error", pSyncIndexMgr->pSyncNode->vgId, host, port, term);
 }
 
 SyncTerm syncIndexMgrGetTerm(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId) {

source/libs/sync/src/syncMain.c

@@ -1282,6 +1282,9 @@ cJSON* syncNode2Json(const SSyncNode* pSyncNode) {
     // snapshot receivers
     cJSON* pReceivers = cJSON_CreateArray();
     cJSON_AddItemToObject(pRoot, "receiver", snapshotReceiver2Json(pSyncNode->pNewNodeReceiver));
+
+    // changing
+    cJSON_AddNumberToObject(pRoot, "changing", pSyncNode->changing);
   }
 
   cJSON* pJson = cJSON_CreateObject();
@@ -1304,26 +1307,31 @@ void syncNodeEventLog(const SSyncNode* pSyncNode, char* str) {
     pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
   }
   SyncIndex logLastIndex = pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore);
+  SyncIndex logBeginIndex = pSyncNode->pLogStore->syncLogBeginIndex(pSyncNode->pLogStore);
 
   if (userStrLen < 256) {
     char logBuf[128 + 256];
     snprintf(logBuf, sizeof(logBuf),
-             "vgId:%d, sync %s %s, term:%lu, commit:%ld, lastlog:%ld, lastsnapshot:%ld, standby:%d, replica-num:%d, "
+             "vgId:%d, sync %s %s, term:%lu, commit:%ld, beginlog:%ld, lastlog:%ld, lastsnapshot:%ld, standby:%d, "
+             "replica-num:%d, "
              "lconfig:%ld, changing:%d",
              pSyncNode->vgId, syncUtilState2String(pSyncNode->state), str, pSyncNode->pRaftStore->currentTerm,
-             pSyncNode->commitIndex, logLastIndex, snapshot.lastApplyIndex, pSyncNode->pRaftCfg->isStandBy,
+             pSyncNode->commitIndex, logBeginIndex, logLastIndex, snapshot.lastApplyIndex,
-             pSyncNode->replicaNum, pSyncNode->pRaftCfg->lastConfigIndex, pSyncNode->changing);
+             pSyncNode->pRaftCfg->isStandBy, pSyncNode->replicaNum, pSyncNode->pRaftCfg->lastConfigIndex,
+             pSyncNode->changing);
     sDebug("%s", logBuf);
 
   } else {
     int   len = 128 + userStrLen;
     char* s = (char*)taosMemoryMalloc(len);
     snprintf(s, len,
-             "vgId:%d, sync %s %s, term:%lu, commit:%ld, lastlog:%ld, lastsnapshot:%ld, standby:%d, replica-num:%d, "
+             "vgId:%d, sync %s %s, term:%lu, commit:%ld, beginlog:%ld, lastlog:%ld, lastsnapshot:%ld, standby:%d, "
+             "replica-num:%d, "
              "lconfig:%ld, changing:%d",
              pSyncNode->vgId, syncUtilState2String(pSyncNode->state), str, pSyncNode->pRaftStore->currentTerm,
-             pSyncNode->commitIndex, logLastIndex, snapshot.lastApplyIndex, pSyncNode->pRaftCfg->isStandBy,
+             pSyncNode->commitIndex, logBeginIndex, logLastIndex, snapshot.lastApplyIndex,
-             pSyncNode->replicaNum, pSyncNode->pRaftCfg->lastConfigIndex, pSyncNode->changing);
+             pSyncNode->pRaftCfg->isStandBy, pSyncNode->replicaNum, pSyncNode->pRaftCfg->lastConfigIndex,
+             pSyncNode->changing);
     sDebug("%s", s);
     taosMemoryFree(s);
   }
@@ -1400,7 +1408,7 @@ void syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* pNewConfig, SyncInde
     pSyncNode->pRaftCfg->isStandBy = 1;  // set standby
   }
 
-  // persist last config index
+  // add last config index
   raftCfgAddConfigIndex(pSyncNode->pRaftCfg, lastConfigChangeIndex);
 
   if (IamInNew) {
@@ -1827,7 +1835,11 @@ SyncIndex syncNodeSyncStartIndex(SSyncNode* pSyncNode) {
 SyncIndex syncNodeGetPreIndex(SSyncNode* pSyncNode, SyncIndex index) {
   ASSERT(index >= SYNC_INDEX_BEGIN);
   SyncIndex syncStartIndex = syncNodeSyncStartIndex(pSyncNode);
-  ASSERT(index <= syncStartIndex);
+
+  if (index > syncStartIndex) {
+    syncNodeLog3("syncNodeGetPreIndex", pSyncNode);
+    ASSERT(0);
+  }
 
   SyncIndex preIndex = index - 1;
   return preIndex;
@@ -1836,7 +1848,11 @@ SyncIndex syncNodeGetPreIndex(SSyncNode* pSyncNode, SyncIndex index) {
 SyncTerm syncNodeGetPreTerm(SSyncNode* pSyncNode, SyncIndex index) {
   ASSERT(index >= SYNC_INDEX_BEGIN);
   SyncIndex syncStartIndex = syncNodeSyncStartIndex(pSyncNode);
-  ASSERT(index <= syncStartIndex);
+
+  if (index > syncStartIndex) {
+    syncNodeLog3("syncNodeGetPreTerm", pSyncNode);
+    ASSERT(0);
+  }
 
   if (index == SYNC_INDEX_BEGIN) {
     return 0;
@@ -1929,6 +1945,12 @@ void syncNodeLog2(char* s, SSyncNode* pObj) {
   }
 }
 
+void syncNodeLog3(char* s, SSyncNode* pObj) {
+  char* serialized = syncNode2Str(pObj);
+  sTraceLong("syncNodeLog3 | len:%lu | %s | %s", strlen(serialized), s, serialized);
+  taosMemoryFree(serialized);
+}
+
 // ------ local funciton ---------
 // enqueue message ----
 static void syncNodeEqPingTimer(void* param, void* tmrId) {

source/libs/sync/src/syncRaftCfg.c

@@ -53,7 +53,12 @@ int32_t raftCfgPersist(SRaftCfg *pRaftCfg) {
 
   char buf[CONFIG_FILE_LEN] = {0};
   memset(buf, 0, sizeof(buf));
-  ASSERT(strlen(s) + 1 <= CONFIG_FILE_LEN);
+
+  if (strlen(s) + 1 > CONFIG_FILE_LEN) {
+    sError("too long config str:%s", s);
+    ASSERT(0);
+  }
+
   snprintf(buf, sizeof(buf), "%s", s);
   int64_t ret = taosWriteFile(pRaftCfg->pFile, buf, sizeof(buf));
   assert(ret == sizeof(buf));

source/libs/sync/src/syncSnapshot.c

@@ -67,38 +67,34 @@ void snapshotSenderDestroy(SSyncSnapshotSender *pSender) {
 bool snapshotSenderIsStart(SSyncSnapshotSender *pSender) { return pSender->start; }
 
 // begin send snapshot (current term, seq begin)
-void snapshotSenderStart(SSyncSnapshotSender *pSender) {
+void snapshotSenderStart(SSyncSnapshotSender *pSender, SSnapshot snapshot, void *pReader) {
   ASSERT(!snapshotSenderIsStart(pSender));
 
   pSender->seq = SYNC_SNAPSHOT_SEQ_BEGIN;
   pSender->ack = SYNC_SNAPSHOT_SEQ_INVALID;
 
+  // open snapshot reader
+  ASSERT(pSender->pReader == NULL);
+  pSender->pReader = pReader;
+  pSender->snapshot = snapshot;
+
+  /*
     // open snapshot reader
     ASSERT(pSender->pReader == NULL);
     int32_t ret = pSender->pSyncNode->pFsm->FpSnapshotStartRead(pSender->pSyncNode->pFsm, &(pSender->pReader));
     ASSERT(ret == 0);
 
+    // get current snapshot info
+    pSender->pSyncNode->pFsm->FpGetSnapshotInfo(pSender->pSyncNode->pFsm, &(pSender->snapshot));
+  */
+
   if (pSender->pCurrentBlock != NULL) {
     taosMemoryFree(pSender->pCurrentBlock);
   }
 
   pSender->blockLen = 0;
 
-  // get current snapshot info
-  pSender->pSyncNode->pFsm->FpGetSnapshotInfo(pSender->pSyncNode->pFsm, &(pSender->snapshot));
-
-  sTrace("snapshotSenderStart lastApplyIndex:%ld, lastApplyTerm:%lu, lastConfigIndex:%ld",
-         pSender->snapshot.lastApplyIndex, pSender->snapshot.lastApplyTerm, pSender->snapshot.lastConfigIndex);
-
   if (pSender->snapshot.lastConfigIndex != SYNC_INDEX_INVALID) {
-    /*
-    SSyncRaftEntry *pEntry = NULL;
-    int32_t code = pSender->pSyncNode->pLogStore->syncLogGetEntry(pSender->pSyncNode->pLogStore,
-                                                                  pSender->snapshot.lastConfigIndex, &pEntry);
-    ASSERT(code == 0);
-    ASSERT(pEntry != NULL);
-    */
-
     SSyncRaftEntry *pEntry =
         pSender->pSyncNode->pLogStore->getEntry(pSender->pSyncNode->pLogStore, pSender->snapshot.lastConfigIndex);
     ASSERT(pEntry != NULL);

source/libs/sync/test/syncReconfigFinishTest.cpp

@@ -60,14 +60,12 @@ SyncReconfigFinish *createMsg() {
   return pMsg;
 }
 
-
 void test1() {
   SyncReconfigFinish *pMsg = createMsg();
   syncReconfigFinishLog2((char *)"test1:", pMsg);
   syncReconfigFinishDestroy(pMsg);
 }
 
-
 void test2() {
   SyncReconfigFinish *pMsg = createMsg();
   uint32_t            len = pMsg->bytes;