Merge branch '3.0' into feature/TD-11463-3.0

include/common/tmsg.h

@@ -2291,6 +2291,10 @@ typedef struct {
   // TODO: other info needed by task
 } SStreamTaskExecReq;
 
+typedef struct {
+  int32_t reserved;
+} SStreamTaskExecRsp;
+
 #pragma pack(pop)
 
 #ifdef __cplusplus

include/common/tmsgdef.h

@@ -200,6 +200,7 @@ enum {
   // Requests handled by SNODE
   TD_NEW_MSG_SEG(TDMT_SND_MSG)
   TD_DEF_MSG_TYPE(TDMT_SND_TASK_DEPLOY, "snode-task-deploy", SStreamTaskDeployReq, SStreamTaskDeployRsp)
+  TD_DEF_MSG_TYPE(TDMT_SND_TASK_EXEC, "snode-task-exec", SStreamTaskExecReq, SStreamTaskExecRsp)
 
 #if defined(TD_MSG_NUMBER_)
   TDMT_MAX

include/libs/transport/trpc.h

@@ -38,11 +38,14 @@ typedef struct SRpcConnInfo {
 
 typedef struct SRpcMsg {
   tmsg_t  msgType;
+  tmsg_t  expectMsgType;
   void *  pCont;
   int     contLen;
   int32_t code;
   void *  handle;   // rpc handle returned to app
   void *  ahandle;  // app handle set by client
+  int     noResp;   // has response or not(default 0 indicate resp);
+
 } SRpcMsg;
 
 typedef struct SRpcInit {

include/util/tencode.h

@@ -402,10 +402,33 @@ static int32_t tDecodeCStrTo(SCoder* pDecoder, char* val) {
   return 0;
 }
 
+static FORCE_INLINE int32_t tDecodeBinaryAlloc(SCoder* pDecoder, void** val, uint64_t* len) {
+  if (tDecodeU64v(pDecoder, len) < 0) return -1;
+
+  if (TD_CODER_CHECK_CAPACITY_FAILED(pDecoder, *len)) return -1;
+  *val = malloc(*len);
+  if (*val == NULL) return -1;
+  memcpy(*val, TD_CODER_CURRENT(pDecoder), *len);
+
+  TD_CODER_MOVE_POS(pDecoder, *len);
+  return 0;
+}
+
+static FORCE_INLINE int32_t tDecodeCStrAndLenAlloc(SCoder* pDecoder, char** val, uint64_t* len) {
+  if (tDecodeBinaryAlloc(pDecoder, (void**)val, len) < 0) return -1;
+  (*len) -= 1;
+  return 0;
+}
+
+static FORCE_INLINE int32_t tDecodeCStrAlloc(SCoder* pDecoder, char** val) {
+  uint64_t len;
+  return tDecodeCStrAndLenAlloc(pDecoder, val, &len);
+}
+
 static FORCE_INLINE bool tDecodeIsEnd(SCoder* pCoder) { return (pCoder->size == pCoder->pos); }
 
 #ifdef __cplusplus
 }
 #endif
 
-#endif /*_TD_UTIL_ENCODE_H_*/
+#endif /*_TD_UTIL_ENCODE_H_*/
\ No newline at end of file

source/common/src/tmsg.c

@@ -2674,9 +2674,9 @@ int32_t tDeserializeSCMCreateStreamReq(void *buf, int32_t bufLen, SCMCreateStrea
   if (tStartDecode(&decoder) < 0) return -1;
   if (tDecodeCStrTo(&decoder, pReq->name) < 0) return -1;
   if (tDecodeI8(&decoder, &pReq->igExists) < 0) return -1;
-  if (tDecodeCStr(&decoder, (const char **)&pReq->sql) < 0) return -1;
-  if (tDecodeCStr(&decoder, (const char **)&pReq->physicalPlan) < 0) return -1;
-  if (tDecodeCStr(&decoder, (const char **)&pReq->logicalPlan) < 0) return -1;
+  if (tDecodeCStrAlloc(&decoder, &pReq->sql) < 0) return -1;
+  if (tDecodeCStrAlloc(&decoder, &pReq->physicalPlan) < 0) return -1;
+  if (tDecodeCStrAlloc(&decoder, &pReq->logicalPlan) < 0) return -1;
   tEndDecode(&decoder);
 
   tCoderClear(&decoder);
@@ -2706,7 +2706,7 @@ int32_t tDecodeSStreamTask(SCoder *pDecoder, SStreamTask *pTask) {
   if (tDecodeI32(pDecoder, &pTask->taskId) < 0) return -1;
   if (tDecodeI32(pDecoder, &pTask->level) < 0) return -1;
   if (tDecodeI8(pDecoder, &pTask->status) < 0) return -1;
-  if (tDecodeCStr(pDecoder, (const char **)&pTask->qmsg) < 0) return -1;
+  if (tDecodeCStrAlloc(pDecoder, &pTask->qmsg) < 0) return -1;
   tEndDecode(pDecoder);
   return 0;
 }

source/dnode/mnode/impl/src/mndDef.c

@@ -39,8 +39,8 @@ int32_t tDecodeSStreamObj(SCoder *pDecoder, SStreamObj *pObj) {
   if (tDecodeI64(pDecoder, &pObj->dbUid) < 0) return -1;
   if (tDecodeI32(pDecoder, &pObj->version) < 0) return -1;
   if (tDecodeI8(pDecoder, &pObj->status) < 0) return -1;
-  if (tDecodeCStr(pDecoder, (const char **)&pObj->sql) < 0) return -1;
-  if (tDecodeCStr(pDecoder, (const char **)&pObj->logicalPlan) < 0) return -1;
-  if (tDecodeCStr(pDecoder, (const char **)&pObj->physicalPlan) < 0) return -1;
+  if (tDecodeCStrAlloc(pDecoder, &pObj->sql) < 0) return -1;
+  if (tDecodeCStrAlloc(pDecoder, &pObj->logicalPlan) < 0) return -1;
+  if (tDecodeCStrAlloc(pDecoder, &pObj->physicalPlan) < 0) return -1;
   return 0;
 }

source/dnode/snode/inc/sndInt.h

@@ -50,8 +50,9 @@ typedef struct SSnode {
 SStreamMeta* sndMetaNew();
 void         sndMetaDelete(SStreamMeta* pMeta);
 
-int32_t sndMetaDeployTask(SStreamMeta* pMeta, SStreamTask* pTask);
-int32_t sndMetaRemoveTask(SStreamMeta* pMeta, int32_t taskId);
+int32_t      sndMetaDeployTask(SStreamMeta* pMeta, SStreamTask* pTask);
+SStreamTask* sndMetaGetTask(SStreamMeta* pMeta, int32_t taskId);
+int32_t      sndMetaRemoveTask(SStreamMeta* pMeta, int32_t taskId);
 
 int32_t sndDropTaskOfStream(SStreamMeta* pMeta, int64_t streamId);
 

source/dnode/snode/src/snode.c

@@ -68,6 +68,10 @@ int32_t sndMetaDeployTask(SStreamMeta *pMeta, SStreamTask *pTask) {
   return taosHashPut(pMeta->pHash, &pTask->taskId, sizeof(int32_t), pTask, sizeof(void *));
 }
 
+SStreamTask *sndMetaGetTask(SStreamMeta *pMeta, int32_t taskId) {
+  return taosHashGet(pMeta->pHash, &taskId, sizeof(int32_t));
+}
+
 int32_t sndMetaRemoveTask(SStreamMeta *pMeta, int32_t taskId) {
   SStreamTask *pTask = taosHashGet(pMeta->pHash, &taskId, sizeof(int32_t));
   if (pTask == NULL) {
@@ -79,6 +83,16 @@ int32_t sndMetaRemoveTask(SStreamMeta *pMeta, int32_t taskId) {
   return taosHashRemove(pMeta->pHash, &taskId, sizeof(int32_t));
 }
 
+static int32_t sndProcessTaskExecReq(SSnode *pSnode, SRpcMsg *pMsg) {
+  SMsgHead    *pHead = pMsg->pCont;
+  int32_t      taskId = pHead->streamTaskId;
+  SStreamTask *pTask = sndMetaGetTask(pSnode->pMeta, taskId);
+  if (pTask == NULL) {
+    return -1;
+  }
+  return 0;
+}
+
 int32_t sndProcessUMsg(SSnode *pSnode, SRpcMsg *pMsg) {
   // stream deploy
   // stream stop/resume
@@ -95,13 +109,20 @@ int32_t sndProcessUMsg(SSnode *pSnode, SRpcMsg *pMsg) {
     tCoderClear(&decoder);
 
     sndMetaDeployTask(pSnode->pMeta, pTask);
+  } else if (pMsg->msgType == TDMT_SND_TASK_EXEC) {
+    sndProcessTaskExecReq(pSnode, pMsg);
   } else {
-    //
+    ASSERT(0);
   }
   return 0;
 }
 
 int32_t sndProcessSMsg(SSnode *pSnode, SRpcMsg *pMsg) {
   // operator exec
+  if (pMsg->msgType == TDMT_SND_TASK_EXEC) {
+    sndProcessTaskExecReq(pSnode, pMsg);
+  } else {
+    ASSERT(0);
+  }
   return 0;
 }

source/libs/sync/inc/syncEnv.h

@@ -31,10 +31,10 @@ extern "C" {
 #define TIMER_MAX_MS 0x7FFFFFFF
 #define ENV_TICK_TIMER_MS 1000
 #define PING_TIMER_MS 1000
-#define ELECT_TIMER_MS_MIN 150
-#define ELECT_TIMER_MS_MAX 300
+#define ELECT_TIMER_MS_MIN 1500
+#define ELECT_TIMER_MS_MAX 3000
 #define ELECT_TIMER_MS_RANGE (ELECT_TIMER_MS_MAX - ELECT_TIMER_MS_MIN)
-#define HEARTBEAT_TIMER_MS 30
+#define HEARTBEAT_TIMER_MS 300
 
 #define EMPTY_RAFT_ID ((SRaftId){.addr = 0, .vgId = 0})
 

source/libs/sync/inc/syncIO.h

@@ -34,11 +34,11 @@ extern "C" {
 
 typedef struct SSyncIO {
   STaosQueue *pMsgQ;
-  STaosQset  *pQset;
+  STaosQset * pQset;
   pthread_t   consumerTid;
 
-  void  *serverRpc;
-  void  *clientRpc;
+  void * serverRpc;
+  void * clientRpc;
   SEpSet myAddr;
 
   tmr_h   qTimer;

source/libs/sync/inc/syncRaftStore.h

@@ -43,11 +43,14 @@ int32_t     raftStorePersist(SRaftStore *pRaftStore);
 int32_t     raftStoreSerialize(SRaftStore *pRaftStore, char *buf, size_t len);
 int32_t     raftStoreDeserialize(SRaftStore *pRaftStore, char *buf, size_t len);
 
-bool raftStoreHasVoted(SRaftStore *pRaftStore);
-void raftStoreVote(SRaftStore *pRaftStore, SRaftId *pRaftId);
-void raftStoreClearVote(SRaftStore *pRaftStore);
-void raftStoreNextTerm(SRaftStore *pRaftStore);
-void raftStoreSetTerm(SRaftStore *pRaftStore, SyncTerm term);
+bool    raftStoreHasVoted(SRaftStore *pRaftStore);
+void    raftStoreVote(SRaftStore *pRaftStore, SRaftId *pRaftId);
+void    raftStoreClearVote(SRaftStore *pRaftStore);
+void    raftStoreNextTerm(SRaftStore *pRaftStore);
+void    raftStoreSetTerm(SRaftStore *pRaftStore, SyncTerm term);
+int32_t raftStoreFromJson(SRaftStore *pRaftStore, cJSON *pJson);
+cJSON * raftStore2Json(SRaftStore *pRaftStore);
+char *  raftStore2Str(SRaftStore *pRaftStore);
 
 // for debug -------------------
 void raftStorePrint(SRaftStore *pObj);

source/libs/sync/src/syncAppendEntries.c

@@ -102,7 +102,7 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
 
   SyncTerm localPreLogTerm = 0;
   if (pMsg->prevLogTerm >= SYNC_INDEX_BEGIN && pMsg->prevLogTerm <= ths->pLogStore->getLastIndex(ths->pLogStore)) {
-    SSyncRaftEntry* pEntry = logStoreGetEntry(ths->pLogStore, pMsg->prevLogTerm);
+    SSyncRaftEntry* pEntry = logStoreGetEntry(ths->pLogStore, pMsg->prevLogIndex);
     assert(pEntry != NULL);
     localPreLogTerm = pEntry->term;
     syncEntryDestory(pEntry);
@@ -111,9 +111,9 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
   bool logOK =
       (pMsg->prevLogIndex == SYNC_INDEX_INVALID) ||
       ((pMsg->prevLogIndex >= SYNC_INDEX_BEGIN) &&
-       (pMsg->prevLogIndex <= ths->pLogStore->getLastIndex(ths->pLogStore)) && (pMsg->prevLogIndex == localPreLogTerm));
+       (pMsg->prevLogIndex <= ths->pLogStore->getLastIndex(ths->pLogStore)) && (pMsg->prevLogTerm == localPreLogTerm));
 
-  // reject
+  // reject request
   if ((pMsg->term < ths->pRaftStore->currentTerm) ||
       ((pMsg->term == ths->pRaftStore->currentTerm) && (ths->state == TAOS_SYNC_STATE_FOLLOWER) && !logOK)) {
     SyncAppendEntriesReply* pReply = syncAppendEntriesReplyBuild();
@@ -134,6 +134,9 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
   // return to follower state
   if (pMsg->term == ths->pRaftStore->currentTerm && ths->state == TAOS_SYNC_STATE_CANDIDATE) {
     syncNodeBecomeFollower(ths);
+
+    // need ret?
+    return ret;
   }
 
   // accept request
@@ -144,17 +147,17 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
       matchSuccess = true;
     }
     if (pMsg->prevLogIndex >= SYNC_INDEX_BEGIN && pMsg->prevLogIndex <= ths->pLogStore->getLastIndex(ths->pLogStore)) {
-      SSyncRaftEntry* pEntry = logStoreGetEntry(ths->pLogStore, pMsg->prevLogTerm);
-      assert(pEntry != NULL);
-      if (pMsg->prevLogTerm == pEntry->term) {
+      SSyncRaftEntry* pPreEntry = logStoreGetEntry(ths->pLogStore, pMsg->prevLogIndex);
+      assert(pPreEntry != NULL);
+      if (pMsg->prevLogTerm == pPreEntry->term) {
         matchSuccess = true;
       }
-      syncEntryDestory(pEntry);
+      syncEntryDestory(pPreEntry);
     }
 
     if (matchSuccess) {
       // delete conflict entries
-      if (ths->pLogStore->getLastIndex(ths->pLogStore) > pMsg->prevLogIndex) {
+      if (pMsg->prevLogIndex < ths->pLogStore->getLastIndex(ths->pLogStore)) {
         SyncIndex fromIndex = pMsg->prevLogIndex + 1;
         ths->pLogStore->truncate(ths->pLogStore, fromIndex);
       }
@@ -178,6 +181,7 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
       syncNodeSendMsgById(&pReply->destId, ths, &rpcMsg);
 
       syncAppendEntriesReplyDestroy(pReply);
+
     } else {
       SyncAppendEntriesReply* pReply = syncAppendEntriesReplyBuild();
       pReply->srcId = ths->myRaftId;

source/libs/sync/src/syncAppendEntriesReply.c

@@ -51,10 +51,10 @@ int32_t syncNodeOnAppendEntriesReplyCb(SSyncNode* ths, SyncAppendEntriesReply* p
   assert(pMsg->term == ths->pRaftStore->currentTerm);
 
   if (pMsg->success) {
-    // nextIndex = reply.matchIndex + 1
+    // nextIndex'  = [nextIndex  EXCEPT ![i][j] = m.mmatchIndex + 1]
     syncIndexMgrSetIndex(ths->pNextIndex, &(pMsg->srcId), pMsg->matchIndex + 1);
 
-    // matchIndex = reply.matchIndex
+    // matchIndex' = [matchIndex EXCEPT ![i][j] = m.mmatchIndex]
     syncIndexMgrSetIndex(ths->pMatchIndex, &(pMsg->srcId), pMsg->matchIndex);
 
     // maybe commit
@@ -62,6 +62,8 @@ int32_t syncNodeOnAppendEntriesReplyCb(SSyncNode* ths, SyncAppendEntriesReply* p
 
   } else {
     SyncIndex nextIndex = syncIndexMgrGetIndex(ths->pNextIndex, &(pMsg->srcId));
+
+    // notice! int64, uint64
     if (nextIndex > SYNC_INDEX_BEGIN) {
       --nextIndex;
     } else {

source/libs/sync/src/syncCommit.c

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "syncIndexMgr.h"
+#include "syncInt.h"
+
+// \* Leader i advances its commitIndex.
+// \* This is done as a separate step from handling AppendEntries responses,
+// \* in part to minimize atomic regions, and in part so that leaders of
+// \* single-server clusters are able to mark entries committed.
+// AdvanceCommitIndex(i) ==
+//     /\ state[i] = Leader
+//     /\ LET \* The set of servers that agree up through index.
+//            Agree(index) == {i} \cup {k \in Server :
+//                                          matchIndex[i][k] >= index}
+//            \* The maximum indexes for which a quorum agrees
+//            agreeIndexes == {index \in 1..Len(log[i]) :
+//                                 Agree(index) \in Quorum}
+//            \* New value for commitIndex'[i]
+//            newCommitIndex ==
+//               IF /\ agreeIndexes /= {}
+//                  /\ log[i][Max(agreeIndexes)].term = currentTerm[i]
+//               THEN
+//                   Max(agreeIndexes)
+//               ELSE
+//                   commitIndex[i]
+//        IN commitIndex' = [commitIndex EXCEPT ![i] = newCommitIndex]
+//     /\ UNCHANGED <<messages, serverVars, candidateVars, leaderVars, log>>
+//
+void syncNodeMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
+  syncIndexMgrLog2("==syncNodeMaybeAdvanceCommitIndex== pNextIndex", pSyncNode->pNextIndex);
+  syncIndexMgrLog2("==syncNodeMaybeAdvanceCommitIndex== pMatchIndex", pSyncNode->pMatchIndex);
+}
\ No newline at end of file

source/libs/sync/src/syncElection.c

@@ -50,6 +50,7 @@ int32_t syncNodeRequestVotePeers(SSyncNode* pSyncNode) {
 }
 
 int32_t syncNodeElect(SSyncNode* pSyncNode) {
+  int32_t ret = 0;
   if (pSyncNode->state == TAOS_SYNC_STATE_FOLLOWER) {
     syncNodeFollower2Candidate(pSyncNode);
   }
@@ -62,7 +63,15 @@ int32_t syncNodeElect(SSyncNode* pSyncNode) {
   votesRespondReset(pSyncNode->pVotesRespond, pSyncNode->pRaftStore->currentTerm);
 
   syncNodeVoteForSelf(pSyncNode);
-  int32_t ret = syncNodeRequestVotePeers(pSyncNode);
+  if (voteGrantedMajority(pSyncNode->pVotesGranted)) {
+    // only myself, to leader
+    assert(!pSyncNode->pVotesGranted->toLeader);
+    syncNodeCandidate2Leader(pSyncNode);
+    pSyncNode->pVotesGranted->toLeader = true;
+    return ret;
+  }
+
+  ret = syncNodeRequestVotePeers(pSyncNode);
   assert(ret == 0);
   syncNodeResetElectTimer(pSyncNode);
 

source/libs/sync/src/syncIO.c

@@ -269,6 +269,7 @@ static void *syncIOConsumerFunc(void *param) {
       } else if (pRpcMsg->msgType == SYNC_PING_REPLY) {
         if (io->FpOnSyncPingReply != NULL) {
           SyncPingReply *pSyncMsg = syncPingReplyFromRpcMsg2(pRpcMsg);
+          assert(pSyncMsg != NULL);
           io->FpOnSyncPingReply(io->pSyncNode, pSyncMsg);
           syncPingReplyDestroy(pSyncMsg);
         }
@@ -276,6 +277,7 @@ static void *syncIOConsumerFunc(void *param) {
       } else if (pRpcMsg->msgType == SYNC_CLIENT_REQUEST) {
         if (io->FpOnSyncClientRequest != NULL) {
           SyncClientRequest *pSyncMsg = syncClientRequestFromRpcMsg2(pRpcMsg);
+          assert(pSyncMsg != NULL);
           io->FpOnSyncClientRequest(io->pSyncNode, pSyncMsg);
           syncClientRequestDestroy(pSyncMsg);
         }
@@ -283,6 +285,7 @@ static void *syncIOConsumerFunc(void *param) {
       } else if (pRpcMsg->msgType == SYNC_REQUEST_VOTE) {
         if (io->FpOnSyncRequestVote != NULL) {
           SyncRequestVote *pSyncMsg = syncRequestVoteFromRpcMsg2(pRpcMsg);
+          assert(pSyncMsg != NULL);
           io->FpOnSyncRequestVote(io->pSyncNode, pSyncMsg);
           syncRequestVoteDestroy(pSyncMsg);
         }
@@ -290,6 +293,7 @@ static void *syncIOConsumerFunc(void *param) {
       } else if (pRpcMsg->msgType == SYNC_REQUEST_VOTE_REPLY) {
         if (io->FpOnSyncRequestVoteReply != NULL) {
           SyncRequestVoteReply *pSyncMsg = syncRequestVoteReplyFromRpcMsg2(pRpcMsg);
+          assert(pSyncMsg != NULL);
           io->FpOnSyncRequestVoteReply(io->pSyncNode, pSyncMsg);
           syncRequestVoteReplyDestroy(pSyncMsg);
         }
@@ -297,6 +301,7 @@ static void *syncIOConsumerFunc(void *param) {
       } else if (pRpcMsg->msgType == SYNC_APPEND_ENTRIES) {
         if (io->FpOnSyncAppendEntries != NULL) {
           SyncAppendEntries *pSyncMsg = syncAppendEntriesFromRpcMsg2(pRpcMsg);
+          assert(pSyncMsg != NULL);
           io->FpOnSyncAppendEntries(io->pSyncNode, pSyncMsg);
           syncAppendEntriesDestroy(pSyncMsg);
         }
@@ -304,6 +309,7 @@ static void *syncIOConsumerFunc(void *param) {
       } else if (pRpcMsg->msgType == SYNC_APPEND_ENTRIES_REPLY) {
         if (io->FpOnSyncAppendEntriesReply != NULL) {
           SyncAppendEntriesReply *pSyncMsg = syncAppendEntriesReplyFromRpcMsg2(pRpcMsg);
+          assert(pSyncMsg != NULL);
           io->FpOnSyncAppendEntriesReply(io->pSyncNode, pSyncMsg);
           syncAppendEntriesReplyDestroy(pSyncMsg);
         }
@@ -311,6 +317,7 @@ static void *syncIOConsumerFunc(void *param) {
       } else if (pRpcMsg->msgType == SYNC_TIMEOUT) {
         if (io->FpOnSyncTimeout != NULL) {
           SyncTimeout *pSyncMsg = syncTimeoutFromRpcMsg2(pRpcMsg);
+          assert(pSyncMsg != NULL);
           io->FpOnSyncTimeout(io->pSyncNode, pSyncMsg);
           syncTimeoutDestroy(pSyncMsg);
         }

source/libs/sync/src/syncIndexMgr.c

@@ -70,22 +70,24 @@ cJSON *syncIndexMgr2Json(SSyncIndexMgr *pSyncIndexMgr) {
   char   u64buf[128];
   cJSON *pRoot = cJSON_CreateObject();
 
-  cJSON_AddNumberToObject(pRoot, "replicaNum", pSyncIndexMgr->replicaNum);
-  cJSON *pReplicas = cJSON_CreateArray();
-  cJSON_AddItemToObject(pRoot, "replicas", pReplicas);
-  for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
-    cJSON_AddItemToArray(pReplicas, syncUtilRaftId2Json(&(*(pSyncIndexMgr->replicas))[i]));
-  }
-  int  respondNum = 0;
-  int *arr = (int *)malloc(sizeof(int) * pSyncIndexMgr->replicaNum);
-  for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
-    arr[i] = pSyncIndexMgr->index[i];
+  if (pSyncIndexMgr != NULL) {
+    cJSON_AddNumberToObject(pRoot, "replicaNum", pSyncIndexMgr->replicaNum);
+    cJSON *pReplicas = cJSON_CreateArray();
+    cJSON_AddItemToObject(pRoot, "replicas", pReplicas);
+    for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
+      cJSON_AddItemToArray(pReplicas, syncUtilRaftId2Json(&(*(pSyncIndexMgr->replicas))[i]));
+    }
+    int  respondNum = 0;
+    int *arr = (int *)malloc(sizeof(int) * pSyncIndexMgr->replicaNum);
+    for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
+      arr[i] = pSyncIndexMgr->index[i];
+    }
+    cJSON *pIndex = cJSON_CreateIntArray(arr, pSyncIndexMgr->replicaNum);
+    free(arr);
+    cJSON_AddItemToObject(pRoot, "index", pIndex);
+    snprintf(u64buf, sizeof(u64buf), "%p", pSyncIndexMgr->pSyncNode);
+    cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf);
   }
-  cJSON *pIndex = cJSON_CreateIntArray(arr, pSyncIndexMgr->replicaNum);
-  free(arr);
-  cJSON_AddItemToObject(pRoot, "index", pIndex);
-  snprintf(u64buf, sizeof(u64buf), "%p", pSyncIndexMgr->pSyncNode);
-  cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf);
 
   cJSON *pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "pSyncIndexMgr", pRoot);

source/libs/sync/src/syncMain.c

@@ -103,6 +103,12 @@ SSyncNode* syncNodeOpen(const SSyncInfo* pSyncInfo) {
   assert(pSyncNode != NULL);
   memset(pSyncNode, 0, sizeof(SSyncNode));
 
+  if (taosMkDir(pSyncInfo->path) != 0) {
+    terrno = TAOS_SYSTEM_ERROR(errno);
+    sError("failed to create dir:%s since %s", pSyncInfo->path, terrstr());
+    return NULL;
+  }
+
   // init by SSyncInfo
   pSyncNode->vgId = pSyncInfo->vgId;
   pSyncNode->syncCfg = pSyncInfo->syncCfg;
@@ -200,6 +206,9 @@ SSyncNode* syncNodeOpen(const SSyncInfo* pSyncInfo) {
   pSyncNode->FpOnAppendEntriesReply = syncNodeOnAppendEntriesReplyCb;
   pSyncNode->FpOnTimeout = syncNodeOnTimeoutCb;
 
+  // start raft
+  syncNodeBecomeFollower(pSyncNode);
+
   return pSyncNode;
 }
 
@@ -355,129 +364,131 @@ cJSON* syncNode2Json(const SSyncNode* pSyncNode) {
   char   u64buf[128];
   cJSON* pRoot = cJSON_CreateObject();
 
-  // init by SSyncInfo
-  cJSON_AddNumberToObject(pRoot, "vgId", pSyncNode->vgId);
-  cJSON_AddStringToObject(pRoot, "path", pSyncNode->path);
-  snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->pWal);
-  cJSON_AddStringToObject(pRoot, "pWal", u64buf);
-
-  snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->rpcClient);
-  cJSON_AddStringToObject(pRoot, "rpcClient", u64buf);
-  snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpSendMsg);
-  cJSON_AddStringToObject(pRoot, "FpSendMsg", u64buf);
-
-  snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->queue);
-  cJSON_AddStringToObject(pRoot, "queue", u64buf);
-  snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpEqMsg);
-  cJSON_AddStringToObject(pRoot, "FpEqMsg", u64buf);
+  if (pSyncNode != NULL) {
+    // init by SSyncInfo
+    cJSON_AddNumberToObject(pRoot, "vgId", pSyncNode->vgId);
+    cJSON_AddStringToObject(pRoot, "path", pSyncNode->path);
+    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->pWal);
+    cJSON_AddStringToObject(pRoot, "pWal", u64buf);
+
+    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->rpcClient);
+    cJSON_AddStringToObject(pRoot, "rpcClient", u64buf);
+    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpSendMsg);
+    cJSON_AddStringToObject(pRoot, "FpSendMsg", u64buf);
+
+    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->queue);
+    cJSON_AddStringToObject(pRoot, "queue", u64buf);
+    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpEqMsg);
+    cJSON_AddStringToObject(pRoot, "FpEqMsg", u64buf);
+
+    // init internal
+    cJSON* pMe = syncUtilNodeInfo2Json(&pSyncNode->myNodeInfo);
+    cJSON_AddItemToObject(pRoot, "myNodeInfo", pMe);
+    cJSON* pRaftId = syncUtilRaftId2Json(&pSyncNode->myRaftId);
+    cJSON_AddItemToObject(pRoot, "myRaftId", pRaftId);
+
+    cJSON_AddNumberToObject(pRoot, "peersNum", pSyncNode->peersNum);
+    cJSON* pPeers = cJSON_CreateArray();
+    cJSON_AddItemToObject(pRoot, "peersNodeInfo", pPeers);
+    for (int i = 0; i < pSyncNode->peersNum; ++i) {
+      cJSON_AddItemToArray(pPeers, syncUtilNodeInfo2Json(&pSyncNode->peersNodeInfo[i]));
+    }
+    cJSON* pPeersId = cJSON_CreateArray();
+    cJSON_AddItemToObject(pRoot, "peersId", pPeersId);
+    for (int i = 0; i < pSyncNode->peersNum; ++i) {
+      cJSON_AddItemToArray(pPeersId, syncUtilRaftId2Json(&pSyncNode->peersId[i]));
+    }
 
-  // init internal
-  cJSON* pMe = syncUtilNodeInfo2Json(&pSyncNode->myNodeInfo);
-  cJSON_AddItemToObject(pRoot, "myNodeInfo", pMe);
-  cJSON* pRaftId = syncUtilRaftId2Json(&pSyncNode->myRaftId);
-  cJSON_AddItemToObject(pRoot, "myRaftId", pRaftId);
-
-  cJSON_AddNumberToObject(pRoot, "peersNum", pSyncNode->peersNum);
-  cJSON* pPeers = cJSON_CreateArray();
-  cJSON_AddItemToObject(pRoot, "peersNodeInfo", pPeers);
-  for (int i = 0; i < pSyncNode->peersNum; ++i) {
-    cJSON_AddItemToArray(pPeers, syncUtilNodeInfo2Json(&pSyncNode->peersNodeInfo[i]));
-  }
-  cJSON* pPeersId = cJSON_CreateArray();
-  cJSON_AddItemToObject(pRoot, "peersId", pPeersId);
-  for (int i = 0; i < pSyncNode->peersNum; ++i) {
-    cJSON_AddItemToArray(pPeersId, syncUtilRaftId2Json(&pSyncNode->peersId[i]));
-  }
+    cJSON_AddNumberToObject(pRoot, "replicaNum", pSyncNode->replicaNum);
+    cJSON* pReplicasId = cJSON_CreateArray();
+    cJSON_AddItemToObject(pRoot, "replicasId", pReplicasId);
+    for (int i = 0; i < pSyncNode->replicaNum; ++i) {
+      cJSON_AddItemToArray(pReplicasId, syncUtilRaftId2Json(&pSyncNode->replicasId[i]));
+    }
 
-  cJSON_AddNumberToObject(pRoot, "replicaNum", pSyncNode->replicaNum);
-  cJSON* pReplicasId = cJSON_CreateArray();
-  cJSON_AddItemToObject(pRoot, "replicasId", pReplicasId);
-  for (int i = 0; i < pSyncNode->replicaNum; ++i) {
-    cJSON_AddItemToArray(pReplicasId, syncUtilRaftId2Json(&pSyncNode->replicasId[i]));
+    // raft algorithm
+    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->pFsm);
+    cJSON_AddStringToObject(pRoot, "pFsm", u64buf);
+    cJSON_AddNumberToObject(pRoot, "quorum", pSyncNode->quorum);
+    cJSON* pLaderCache = syncUtilRaftId2Json(&pSyncNode->leaderCache);
+    cJSON_AddItemToObject(pRoot, "leaderCache", pLaderCache);
+
+    // tla+ server vars
+    cJSON_AddNumberToObject(pRoot, "state", pSyncNode->state);
+    cJSON_AddStringToObject(pRoot, "state_str", syncUtilState2String(pSyncNode->state));
+    char tmpBuf[RAFT_STORE_BLOCK_SIZE];
+    raftStoreSerialize(pSyncNode->pRaftStore, tmpBuf, sizeof(tmpBuf));
+    cJSON_AddStringToObject(pRoot, "pRaftStore", tmpBuf);
+
+    // tla+ candidate vars
+    cJSON_AddItemToObject(pRoot, "pVotesGranted", voteGranted2Json(pSyncNode->pVotesGranted));
+    cJSON_AddItemToObject(pRoot, "pVotesRespond", votesRespond2Json(pSyncNode->pVotesRespond));
+
+    // tla+ leader vars
+    cJSON_AddItemToObject(pRoot, "pNextIndex", syncIndexMgr2Json(pSyncNode->pNextIndex));
+    cJSON_AddItemToObject(pRoot, "pMatchIndex", syncIndexMgr2Json(pSyncNode->pMatchIndex));
+
+    // tla+ log vars
+    cJSON_AddItemToObject(pRoot, "pLogStore", logStore2Json(pSyncNode->pLogStore));
+    snprintf(u64buf, sizeof(u64buf), "%" PRId64 "", pSyncNode->commitIndex);
+    cJSON_AddStringToObject(pRoot, "commitIndex", u64buf);
+
+    // ping timer
+    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->pPingTimer);
+    cJSON_AddStringToObject(pRoot, "pPingTimer", u64buf);
+    cJSON_AddNumberToObject(pRoot, "pingTimerMS", pSyncNode->pingTimerMS);
+    snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->pingTimerLogicClock);
+    cJSON_AddStringToObject(pRoot, "pingTimerLogicClock", u64buf);
+    snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->pingTimerLogicClockUser);
+    cJSON_AddStringToObject(pRoot, "pingTimerLogicClockUser", u64buf);
+    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpPingTimerCB);
+    cJSON_AddStringToObject(pRoot, "FpPingTimerCB", u64buf);
+    snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->pingTimerCounter);
+    cJSON_AddStringToObject(pRoot, "pingTimerCounter", u64buf);
+
+    // elect timer
+    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->pElectTimer);
+    cJSON_AddStringToObject(pRoot, "pElectTimer", u64buf);
+    cJSON_AddNumberToObject(pRoot, "electTimerMS", pSyncNode->electTimerMS);
+    snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->electTimerLogicClock);
+    cJSON_AddStringToObject(pRoot, "electTimerLogicClock", u64buf);
+    snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->electTimerLogicClockUser);
+    cJSON_AddStringToObject(pRoot, "electTimerLogicClockUser", u64buf);
+    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpElectTimerCB);
+    cJSON_AddStringToObject(pRoot, "FpElectTimerCB", u64buf);
+    snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->electTimerCounter);
+    cJSON_AddStringToObject(pRoot, "electTimerCounter", u64buf);
+
+    // heartbeat timer
+    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->pHeartbeatTimer);
+    cJSON_AddStringToObject(pRoot, "pHeartbeatTimer", u64buf);
+    cJSON_AddNumberToObject(pRoot, "heartbeatTimerMS", pSyncNode->heartbeatTimerMS);
+    snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->heartbeatTimerLogicClock);
+    cJSON_AddStringToObject(pRoot, "heartbeatTimerLogicClock", u64buf);
+    snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->heartbeatTimerLogicClockUser);
+    cJSON_AddStringToObject(pRoot, "heartbeatTimerLogicClockUser", u64buf);
+    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpHeartbeatTimerCB);
+    cJSON_AddStringToObject(pRoot, "FpHeartbeatTimerCB", u64buf);
+    snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->heartbeatTimerCounter);
+    cJSON_AddStringToObject(pRoot, "heartbeatTimerCounter", u64buf);
+
+    // callback
+    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnPing);
+    cJSON_AddStringToObject(pRoot, "FpOnPing", u64buf);
+    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnPingReply);
+    cJSON_AddStringToObject(pRoot, "FpOnPingReply", u64buf);
+    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnRequestVote);
+    cJSON_AddStringToObject(pRoot, "FpOnRequestVote", u64buf);
+    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnRequestVoteReply);
+    cJSON_AddStringToObject(pRoot, "FpOnRequestVoteReply", u64buf);
+    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnAppendEntries);
+    cJSON_AddStringToObject(pRoot, "FpOnAppendEntries", u64buf);
+    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnAppendEntriesReply);
+    cJSON_AddStringToObject(pRoot, "FpOnAppendEntriesReply", u64buf);
+    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnTimeout);
+    cJSON_AddStringToObject(pRoot, "FpOnTimeout", u64buf);
   }
 
-  // raft algorithm
-  snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->pFsm);
-  cJSON_AddStringToObject(pRoot, "pFsm", u64buf);
-  cJSON_AddNumberToObject(pRoot, "quorum", pSyncNode->quorum);
-  cJSON* pLaderCache = syncUtilRaftId2Json(&pSyncNode->leaderCache);
-  cJSON_AddItemToObject(pRoot, "leaderCache", pLaderCache);
-
-  // tla+ server vars
-  cJSON_AddNumberToObject(pRoot, "state", pSyncNode->state);
-  cJSON_AddStringToObject(pRoot, "state_str", syncUtilState2String(pSyncNode->state));
-  char tmpBuf[RAFT_STORE_BLOCK_SIZE];
-  raftStoreSerialize(pSyncNode->pRaftStore, tmpBuf, sizeof(tmpBuf));
-  cJSON_AddStringToObject(pRoot, "pRaftStore", tmpBuf);
-
-  // tla+ candidate vars
-  cJSON_AddItemToObject(pRoot, "pVotesGranted", voteGranted2Json(pSyncNode->pVotesGranted));
-  cJSON_AddItemToObject(pRoot, "pVotesRespond", votesRespond2Json(pSyncNode->pVotesRespond));
-
-  // tla+ leader vars
-  cJSON_AddItemToObject(pRoot, "pNextIndex", syncIndexMgr2Json(pSyncNode->pNextIndex));
-  cJSON_AddItemToObject(pRoot, "pMatchIndex", syncIndexMgr2Json(pSyncNode->pMatchIndex));
-
-  // tla+ log vars
-  cJSON_AddItemToObject(pRoot, "pLogStore", logStore2Json(pSyncNode->pLogStore));
-  snprintf(u64buf, sizeof(u64buf), "%" PRId64 "", pSyncNode->commitIndex);
-  cJSON_AddStringToObject(pRoot, "commitIndex", u64buf);
-
-  // ping timer
-  snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->pPingTimer);
-  cJSON_AddStringToObject(pRoot, "pPingTimer", u64buf);
-  cJSON_AddNumberToObject(pRoot, "pingTimerMS", pSyncNode->pingTimerMS);
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->pingTimerLogicClock);
-  cJSON_AddStringToObject(pRoot, "pingTimerLogicClock", u64buf);
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->pingTimerLogicClockUser);
-  cJSON_AddStringToObject(pRoot, "pingTimerLogicClockUser", u64buf);
-  snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpPingTimerCB);
-  cJSON_AddStringToObject(pRoot, "FpPingTimerCB", u64buf);
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->pingTimerCounter);
-  cJSON_AddStringToObject(pRoot, "pingTimerCounter", u64buf);
-
-  // elect timer
-  snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->pElectTimer);
-  cJSON_AddStringToObject(pRoot, "pElectTimer", u64buf);
-  cJSON_AddNumberToObject(pRoot, "electTimerMS", pSyncNode->electTimerMS);
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->electTimerLogicClock);
-  cJSON_AddStringToObject(pRoot, "electTimerLogicClock", u64buf);
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->electTimerLogicClockUser);
-  cJSON_AddStringToObject(pRoot, "electTimerLogicClockUser", u64buf);
-  snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpElectTimerCB);
-  cJSON_AddStringToObject(pRoot, "FpElectTimerCB", u64buf);
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->electTimerCounter);
-  cJSON_AddStringToObject(pRoot, "electTimerCounter", u64buf);
-
-  // heartbeat timer
-  snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->pHeartbeatTimer);
-  cJSON_AddStringToObject(pRoot, "pHeartbeatTimer", u64buf);
-  cJSON_AddNumberToObject(pRoot, "heartbeatTimerMS", pSyncNode->heartbeatTimerMS);
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->heartbeatTimerLogicClock);
-  cJSON_AddStringToObject(pRoot, "heartbeatTimerLogicClock", u64buf);
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->heartbeatTimerLogicClockUser);
-  cJSON_AddStringToObject(pRoot, "heartbeatTimerLogicClockUser", u64buf);
-  snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpHeartbeatTimerCB);
-  cJSON_AddStringToObject(pRoot, "FpHeartbeatTimerCB", u64buf);
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->heartbeatTimerCounter);
-  cJSON_AddStringToObject(pRoot, "heartbeatTimerCounter", u64buf);
-
-  // callback
-  snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnPing);
-  cJSON_AddStringToObject(pRoot, "FpOnPing", u64buf);
-  snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnPingReply);
-  cJSON_AddStringToObject(pRoot, "FpOnPingReply", u64buf);
-  snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnRequestVote);
-  cJSON_AddStringToObject(pRoot, "FpOnRequestVote", u64buf);
-  snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnRequestVoteReply);
-  cJSON_AddStringToObject(pRoot, "FpOnRequestVoteReply", u64buf);
-  snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnAppendEntries);
-  cJSON_AddStringToObject(pRoot, "FpOnAppendEntries", u64buf);
-  snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnAppendEntriesReply);
-  cJSON_AddStringToObject(pRoot, "FpOnAppendEntriesReply", u64buf);
-  snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnTimeout);
-  cJSON_AddStringToObject(pRoot, "FpOnTimeout", u64buf);
-
   cJSON* pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SSyncNode", pRoot);
   return pJson;
@@ -500,15 +511,17 @@ void syncNodeUpdateTerm(SSyncNode* pSyncNode, SyncTerm term) {
 }
 
 void syncNodeBecomeFollower(SSyncNode* pSyncNode) {
+  // maybe clear leader cache
   if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
     pSyncNode->leaderCache = EMPTY_RAFT_ID;
   }
 
+  // state change
   pSyncNode->state = TAOS_SYNC_STATE_FOLLOWER;
   syncNodeStopHeartbeatTimer(pSyncNode);
 
-  int32_t electMS = syncUtilElectRandomMS();
-  syncNodeRestartElectTimer(pSyncNode, electMS);
+  // reset elect timer
+  syncNodeResetElectTimer(pSyncNode);
 }
 
 // TLA+ Spec
@@ -530,20 +543,32 @@ void syncNodeBecomeFollower(SSyncNode* pSyncNode) {
 //     /\ UNCHANGED <<messages, currentTerm, votedFor, candidateVars, logVars>>
 //
 void syncNodeBecomeLeader(SSyncNode* pSyncNode) {
+  // state change
   pSyncNode->state = TAOS_SYNC_STATE_LEADER;
+
+  // set leader cache
   pSyncNode->leaderCache = pSyncNode->myRaftId;
 
   for (int i = 0; i < pSyncNode->pNextIndex->replicaNum; ++i) {
+    // maybe overwrite myself, no harm
+    // just do it!
     pSyncNode->pNextIndex->index[i] = pSyncNode->pLogStore->getLastIndex(pSyncNode->pLogStore) + 1;
   }
 
   for (int i = 0; i < pSyncNode->pMatchIndex->replicaNum; ++i) {
+    // maybe overwrite myself, no harm
+    // just do it!
     pSyncNode->pMatchIndex->index[i] = SYNC_INDEX_INVALID;
   }
 
+  // stop elect timer
   syncNodeStopElectTimer(pSyncNode);
-  syncNodeStartHeartbeatTimer(pSyncNode);
+
+  // start replicate right now!
   syncNodeReplicate(pSyncNode);
+
+  // start heartbeat timer
+  syncNodeStartHeartbeatTimer(pSyncNode);
 }
 
 void syncNodeCandidate2Leader(SSyncNode* pSyncNode) {
@@ -568,6 +593,9 @@ void syncNodeCandidate2Follower(SSyncNode* pSyncNode) {
 }
 
 // raft vote --------------
+
+// just called by syncNodeVoteForSelf
+// need assert
 void syncNodeVoteForTerm(SSyncNode* pSyncNode, SyncTerm term, SRaftId* pRaftId) {
   assert(term == pSyncNode->pRaftStore->currentTerm);
   assert(!raftStoreHasVoted(pSyncNode->pRaftStore));
@@ -575,6 +603,7 @@ void syncNodeVoteForTerm(SSyncNode* pSyncNode, SyncTerm term, SRaftId* pRaftId)
   raftStoreVote(pSyncNode->pRaftStore, pRaftId);
 }
 
+// simulate get vote from outside
 void syncNodeVoteForSelf(SSyncNode* pSyncNode) {
   syncNodeVoteForTerm(pSyncNode, pSyncNode->pRaftStore->currentTerm, &(pSyncNode->myRaftId));
 
@@ -589,8 +618,6 @@ void syncNodeVoteForSelf(SSyncNode* pSyncNode) {
   syncRequestVoteReplyDestroy(pMsg);
 }
 
-void syncNodeMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {}
-
 // for debug --------------
 void syncNodePrint(SSyncNode* pObj) {
   char* serialized = syncNode2Str(pObj);
@@ -676,7 +703,8 @@ static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) {
     taosTmrReset(syncNodeEqHeartbeatTimer, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager,
                  &pSyncNode->pHeartbeatTimer);
   } else {
-    sTrace("==syncNodeEqHeartbeatTimer== heartbeatTimerLogicClock:%" PRIu64 ", heartbeatTimerLogicClockUser:%" PRIu64 "",
+    sTrace("==syncNodeEqHeartbeatTimer== heartbeatTimerLogicClock:%" PRIu64 ", heartbeatTimerLogicClockUser:%" PRIu64
+           "",
            pSyncNode->heartbeatTimerLogicClock, pSyncNode->heartbeatTimerLogicClockUser);
   }
 }

source/libs/sync/src/syncMessage.c

@@ -212,17 +212,19 @@ SyncTimeout* syncTimeoutFromRpcMsg2(const SRpcMsg* pRpcMsg) {
 }
 
 cJSON* syncTimeout2Json(const SyncTimeout* pMsg) {
-  char u64buf[128];
-
+  char   u64buf[128];
   cJSON* pRoot = cJSON_CreateObject();
-  cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
-  cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
-  cJSON_AddNumberToObject(pRoot, "timeoutType", pMsg->timeoutType);
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pMsg->logicClock);
-  cJSON_AddStringToObject(pRoot, "logicClock", u64buf);
-  cJSON_AddNumberToObject(pRoot, "timerMS", pMsg->timerMS);
-  snprintf(u64buf, sizeof(u64buf), "%p", pMsg->data);
-  cJSON_AddStringToObject(pRoot, "data", u64buf);
+
+  if (pMsg != NULL) {
+    cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
+    cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
+    cJSON_AddNumberToObject(pRoot, "timeoutType", pMsg->timeoutType);
+    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->logicClock);
+    cJSON_AddStringToObject(pRoot, "logicClock", u64buf);
+    cJSON_AddNumberToObject(pRoot, "timerMS", pMsg->timerMS);
+    snprintf(u64buf, sizeof(u64buf), "%p", pMsg->data);
+    cJSON_AddStringToObject(pRoot, "data", u64buf);
+  }
 
   cJSON* pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SyncTimeout", pRoot);
@@ -342,50 +344,52 @@ SyncPing* syncPingFromRpcMsg2(const SRpcMsg* pRpcMsg) {
 }
 
 cJSON* syncPing2Json(const SyncPing* pMsg) {
-  char u64buf[128];
-
+  char   u64buf[128];
   cJSON* pRoot = cJSON_CreateObject();
-  cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
-  cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
-
-  cJSON* pSrcId = cJSON_CreateObject();
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pMsg->srcId.addr);
-  cJSON_AddStringToObject(pSrcId, "addr", u64buf);
-  {
-    uint64_t u64 = pMsg->srcId.addr;
-    cJSON*   pTmp = pSrcId;
-    char     host[128];
-    uint16_t port;
-    syncUtilU642Addr(u64, host, sizeof(host), &port);
-    cJSON_AddStringToObject(pTmp, "addr_host", host);
-    cJSON_AddNumberToObject(pTmp, "addr_port", port);
-  }
-  cJSON_AddNumberToObject(pSrcId, "vgId", pMsg->srcId.vgId);
-  cJSON_AddItemToObject(pRoot, "srcId", pSrcId);
-
-  cJSON* pDestId = cJSON_CreateObject();
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pMsg->destId.addr);
-  cJSON_AddStringToObject(pDestId, "addr", u64buf);
-  {
-    uint64_t u64 = pMsg->destId.addr;
-    cJSON*   pTmp = pDestId;
-    char     host[128];
-    uint16_t port;
-    syncUtilU642Addr(u64, host, sizeof(host), &port);
-    cJSON_AddStringToObject(pTmp, "addr_host", host);
-    cJSON_AddNumberToObject(pTmp, "addr_port", port);
+
+  if (pMsg != NULL) {
+    cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
+    cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
+
+    cJSON* pSrcId = cJSON_CreateObject();
+    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->srcId.addr);
+    cJSON_AddStringToObject(pSrcId, "addr", u64buf);
+    {
+      uint64_t u64 = pMsg->srcId.addr;
+      cJSON*   pTmp = pSrcId;
+      char     host[128];
+      uint16_t port;
+      syncUtilU642Addr(u64, host, sizeof(host), &port);
+      cJSON_AddStringToObject(pTmp, "addr_host", host);
+      cJSON_AddNumberToObject(pTmp, "addr_port", port);
+    }
+    cJSON_AddNumberToObject(pSrcId, "vgId", pMsg->srcId.vgId);
+    cJSON_AddItemToObject(pRoot, "srcId", pSrcId);
+
+    cJSON* pDestId = cJSON_CreateObject();
+    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->destId.addr);
+    cJSON_AddStringToObject(pDestId, "addr", u64buf);
+    {
+      uint64_t u64 = pMsg->destId.addr;
+      cJSON*   pTmp = pDestId;
+      char     host[128];
+      uint16_t port;
+      syncUtilU642Addr(u64, host, sizeof(host), &port);
+      cJSON_AddStringToObject(pTmp, "addr_host", host);
+      cJSON_AddNumberToObject(pTmp, "addr_port", port);
+    }
+    cJSON_AddNumberToObject(pDestId, "vgId", pMsg->destId.vgId);
+    cJSON_AddItemToObject(pRoot, "destId", pDestId);
+
+    cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen);
+    char* s;
+    s = syncUtilprintBin((char*)(pMsg->data), pMsg->dataLen);
+    cJSON_AddStringToObject(pRoot, "data", s);
+    free(s);
+    s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
+    cJSON_AddStringToObject(pRoot, "data2", s);
+    free(s);
   }
-  cJSON_AddNumberToObject(pDestId, "vgId", pMsg->destId.vgId);
-  cJSON_AddItemToObject(pRoot, "destId", pDestId);
-
-  cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen);
-  char* s;
-  s = syncUtilprintBin((char*)(pMsg->data), pMsg->dataLen);
-  cJSON_AddStringToObject(pRoot, "data", s);
-  free(s);
-  s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
-  cJSON_AddStringToObject(pRoot, "data2", s);
-  free(s);
 
   cJSON* pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SyncPing", pRoot);
@@ -505,50 +509,52 @@ SyncPingReply* syncPingReplyFromRpcMsg2(const SRpcMsg* pRpcMsg) {
 }
 
 cJSON* syncPingReply2Json(const SyncPingReply* pMsg) {
-  char u64buf[128];
-
+  char   u64buf[128];
   cJSON* pRoot = cJSON_CreateObject();
-  cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
-  cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
-
-  cJSON* pSrcId = cJSON_CreateObject();
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pMsg->srcId.addr);
-  cJSON_AddStringToObject(pSrcId, "addr", u64buf);
-  {
-    uint64_t u64 = pMsg->srcId.addr;
-    cJSON*   pTmp = pSrcId;
-    char     host[128];
-    uint16_t port;
-    syncUtilU642Addr(u64, host, sizeof(host), &port);
-    cJSON_AddStringToObject(pTmp, "addr_host", host);
-    cJSON_AddNumberToObject(pTmp, "addr_port", port);
-  }
-  cJSON_AddNumberToObject(pSrcId, "vgId", pMsg->srcId.vgId);
-  cJSON_AddItemToObject(pRoot, "srcId", pSrcId);
-
-  cJSON* pDestId = cJSON_CreateObject();
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pMsg->destId.addr);
-  cJSON_AddStringToObject(pDestId, "addr", u64buf);
-  {
-    uint64_t u64 = pMsg->destId.addr;
-    cJSON*   pTmp = pDestId;
-    char     host[128];
-    uint16_t port;
-    syncUtilU642Addr(u64, host, sizeof(host), &port);
-    cJSON_AddStringToObject(pTmp, "addr_host", host);
-    cJSON_AddNumberToObject(pTmp, "addr_port", port);
+
+  if (pMsg != NULL) {
+    cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
+    cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
+
+    cJSON* pSrcId = cJSON_CreateObject();
+    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->srcId.addr);
+    cJSON_AddStringToObject(pSrcId, "addr", u64buf);
+    {
+      uint64_t u64 = pMsg->srcId.addr;
+      cJSON*   pTmp = pSrcId;
+      char     host[128];
+      uint16_t port;
+      syncUtilU642Addr(u64, host, sizeof(host), &port);
+      cJSON_AddStringToObject(pTmp, "addr_host", host);
+      cJSON_AddNumberToObject(pTmp, "addr_port", port);
+    }
+    cJSON_AddNumberToObject(pSrcId, "vgId", pMsg->srcId.vgId);
+    cJSON_AddItemToObject(pRoot, "srcId", pSrcId);
+
+    cJSON* pDestId = cJSON_CreateObject();
+    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->destId.addr);
+    cJSON_AddStringToObject(pDestId, "addr", u64buf);
+    {
+      uint64_t u64 = pMsg->destId.addr;
+      cJSON*   pTmp = pDestId;
+      char     host[128];
+      uint16_t port;
+      syncUtilU642Addr(u64, host, sizeof(host), &port);
+      cJSON_AddStringToObject(pTmp, "addr_host", host);
+      cJSON_AddNumberToObject(pTmp, "addr_port", port);
+    }
+    cJSON_AddNumberToObject(pDestId, "vgId", pMsg->destId.vgId);
+    cJSON_AddItemToObject(pRoot, "destId", pDestId);
+
+    cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen);
+    char* s;
+    s = syncUtilprintBin((char*)(pMsg->data), pMsg->dataLen);
+    cJSON_AddStringToObject(pRoot, "data", s);
+    free(s);
+    s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
+    cJSON_AddStringToObject(pRoot, "data2", s);
+    free(s);
   }
-  cJSON_AddNumberToObject(pDestId, "vgId", pMsg->destId.vgId);
-  cJSON_AddItemToObject(pRoot, "destId", pDestId);
-
-  cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen);
-  char* s;
-  s = syncUtilprintBin((char*)(pMsg->data), pMsg->dataLen);
-  cJSON_AddStringToObject(pRoot, "data", s);
-  free(s);
-  s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
-  cJSON_AddStringToObject(pRoot, "data2", s);
-  free(s);
 
   cJSON* pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SyncPingReply", pRoot);
@@ -664,24 +670,26 @@ SyncClientRequest* syncClientRequestFromRpcMsg2(const SRpcMsg* pRpcMsg) {
 }
 
 cJSON* syncClientRequest2Json(const SyncClientRequest* pMsg) {
-  char u64buf[128];
-
+  char   u64buf[128];
   cJSON* pRoot = cJSON_CreateObject();
-  cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
-  cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
-  cJSON_AddNumberToObject(pRoot, "originalRpcType", pMsg->originalRpcType);
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pMsg->seqNum);
-  cJSON_AddStringToObject(pRoot, "seqNum", u64buf);
-  cJSON_AddNumberToObject(pRoot, "isWeak", pMsg->isWeak);
-  cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen);
-
-  char* s;
-  s = syncUtilprintBin((char*)(pMsg->data), pMsg->dataLen);
-  cJSON_AddStringToObject(pRoot, "data", s);
-  free(s);
-  s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
-  cJSON_AddStringToObject(pRoot, "data2", s);
-  free(s);
+
+  if (pMsg != NULL) {
+    cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
+    cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
+    cJSON_AddNumberToObject(pRoot, "originalRpcType", pMsg->originalRpcType);
+    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->seqNum);
+    cJSON_AddStringToObject(pRoot, "seqNum", u64buf);
+    cJSON_AddNumberToObject(pRoot, "isWeak", pMsg->isWeak);
+    cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen);
+
+    char* s;
+    s = syncUtilprintBin((char*)(pMsg->data), pMsg->dataLen);
+    cJSON_AddStringToObject(pRoot, "data", s);
+    free(s);
+    s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
+    cJSON_AddStringToObject(pRoot, "data2", s);
+    free(s);
+  }
 
   cJSON* pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SyncClientRequest", pRoot);
@@ -785,47 +793,49 @@ SyncRequestVote* syncRequestVoteFromRpcMsg2(const SRpcMsg* pRpcMsg) {
 }
 
 cJSON* syncRequestVote2Json(const SyncRequestVote* pMsg) {
-  char u64buf[128];
-
+  char   u64buf[128];
   cJSON* pRoot = cJSON_CreateObject();
-  cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
-  cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
-
-  cJSON* pSrcId = cJSON_CreateObject();
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pMsg->srcId.addr);
-  cJSON_AddStringToObject(pSrcId, "addr", u64buf);
-  {
-    uint64_t u64 = pMsg->srcId.addr;
-    cJSON*   pTmp = pSrcId;
-    char     host[128];
-    uint16_t port;
-    syncUtilU642Addr(u64, host, sizeof(host), &port);
-    cJSON_AddStringToObject(pTmp, "addr_host", host);
-    cJSON_AddNumberToObject(pTmp, "addr_port", port);
-  }
-  cJSON_AddNumberToObject(pSrcId, "vgId", pMsg->srcId.vgId);
-  cJSON_AddItemToObject(pRoot, "srcId", pSrcId);
-
-  cJSON* pDestId = cJSON_CreateObject();
-  cJSON_AddNumberToObject(pDestId, "addr", pMsg->destId.addr);
-  {
-    uint64_t u64 = pMsg->destId.addr;
-    cJSON*   pTmp = pDestId;
-    char     host[128];
-    uint16_t port;
-    syncUtilU642Addr(u64, host, sizeof(host), &port);
-    cJSON_AddStringToObject(pTmp, "addr_host", host);
-    cJSON_AddNumberToObject(pTmp, "addr_port", port);
-  }
-  cJSON_AddNumberToObject(pDestId, "vgId", pMsg->destId.vgId);
-  cJSON_AddItemToObject(pRoot, "destId", pDestId);
 
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pMsg->term);
-  cJSON_AddStringToObject(pRoot, "term", u64buf);
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pMsg->lastLogIndex);
-  cJSON_AddStringToObject(pRoot, "lastLogIndex", u64buf);
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pMsg->lastLogTerm);
-  cJSON_AddStringToObject(pRoot, "lastLogTerm", u64buf);
+  if (pMsg != NULL) {
+    cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
+    cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
+
+    cJSON* pSrcId = cJSON_CreateObject();
+    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->srcId.addr);
+    cJSON_AddStringToObject(pSrcId, "addr", u64buf);
+    {
+      uint64_t u64 = pMsg->srcId.addr;
+      cJSON*   pTmp = pSrcId;
+      char     host[128];
+      uint16_t port;
+      syncUtilU642Addr(u64, host, sizeof(host), &port);
+      cJSON_AddStringToObject(pTmp, "addr_host", host);
+      cJSON_AddNumberToObject(pTmp, "addr_port", port);
+    }
+    cJSON_AddNumberToObject(pSrcId, "vgId", pMsg->srcId.vgId);
+    cJSON_AddItemToObject(pRoot, "srcId", pSrcId);
+
+    cJSON* pDestId = cJSON_CreateObject();
+    cJSON_AddNumberToObject(pDestId, "addr", pMsg->destId.addr);
+    {
+      uint64_t u64 = pMsg->destId.addr;
+      cJSON*   pTmp = pDestId;
+      char     host[128];
+      uint16_t port;
+      syncUtilU642Addr(u64, host, sizeof(host), &port);
+      cJSON_AddStringToObject(pTmp, "addr_host", host);
+      cJSON_AddNumberToObject(pTmp, "addr_port", port);
+    }
+    cJSON_AddNumberToObject(pDestId, "vgId", pMsg->destId.vgId);
+    cJSON_AddItemToObject(pRoot, "destId", pDestId);
+
+    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->term);
+    cJSON_AddStringToObject(pRoot, "term", u64buf);
+    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->lastLogIndex);
+    cJSON_AddStringToObject(pRoot, "lastLogIndex", u64buf);
+    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->lastLogTerm);
+    cJSON_AddStringToObject(pRoot, "lastLogTerm", u64buf);
+  }
 
   cJSON* pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SyncRequestVote", pRoot);
@@ -929,44 +939,46 @@ SyncRequestVoteReply* syncRequestVoteReplyFromRpcMsg2(const SRpcMsg* pRpcMsg) {
 }
 
 cJSON* syncRequestVoteReply2Json(const SyncRequestVoteReply* pMsg) {
-  char u64buf[128];
-
+  char   u64buf[128];
   cJSON* pRoot = cJSON_CreateObject();
-  cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
-  cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
-
-  cJSON* pSrcId = cJSON_CreateObject();
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pMsg->srcId.addr);
-  cJSON_AddStringToObject(pSrcId, "addr", u64buf);
-  {
-    uint64_t u64 = pMsg->srcId.addr;
-    cJSON*   pTmp = pSrcId;
-    char     host[128];
-    uint16_t port;
-    syncUtilU642Addr(u64, host, sizeof(host), &port);
-    cJSON_AddStringToObject(pTmp, "addr_host", host);
-    cJSON_AddNumberToObject(pTmp, "addr_port", port);
-  }
-  cJSON_AddNumberToObject(pSrcId, "vgId", pMsg->srcId.vgId);
-  cJSON_AddItemToObject(pRoot, "srcId", pSrcId);
-
-  cJSON* pDestId = cJSON_CreateObject();
-  cJSON_AddNumberToObject(pDestId, "addr", pMsg->destId.addr);
-  {
-    uint64_t u64 = pMsg->destId.addr;
-    cJSON*   pTmp = pDestId;
-    char     host[128];
-    uint16_t port;
-    syncUtilU642Addr(u64, host, sizeof(host), &port);
-    cJSON_AddStringToObject(pTmp, "addr_host", host);
-    cJSON_AddNumberToObject(pTmp, "addr_port", port);
-  }
-  cJSON_AddNumberToObject(pDestId, "vgId", pMsg->destId.vgId);
-  cJSON_AddItemToObject(pRoot, "destId", pDestId);
 
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pMsg->term);
-  cJSON_AddStringToObject(pRoot, "term", u64buf);
-  cJSON_AddNumberToObject(pRoot, "vote_granted", pMsg->voteGranted);
+  if (pMsg != NULL) {
+    cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
+    cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
+
+    cJSON* pSrcId = cJSON_CreateObject();
+    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->srcId.addr);
+    cJSON_AddStringToObject(pSrcId, "addr", u64buf);
+    {
+      uint64_t u64 = pMsg->srcId.addr;
+      cJSON*   pTmp = pSrcId;
+      char     host[128];
+      uint16_t port;
+      syncUtilU642Addr(u64, host, sizeof(host), &port);
+      cJSON_AddStringToObject(pTmp, "addr_host", host);
+      cJSON_AddNumberToObject(pTmp, "addr_port", port);
+    }
+    cJSON_AddNumberToObject(pSrcId, "vgId", pMsg->srcId.vgId);
+    cJSON_AddItemToObject(pRoot, "srcId", pSrcId);
+
+    cJSON* pDestId = cJSON_CreateObject();
+    cJSON_AddNumberToObject(pDestId, "addr", pMsg->destId.addr);
+    {
+      uint64_t u64 = pMsg->destId.addr;
+      cJSON*   pTmp = pDestId;
+      char     host[128];
+      uint16_t port;
+      syncUtilU642Addr(u64, host, sizeof(host), &port);
+      cJSON_AddStringToObject(pTmp, "addr_host", host);
+      cJSON_AddNumberToObject(pTmp, "addr_port", port);
+    }
+    cJSON_AddNumberToObject(pDestId, "vgId", pMsg->destId.vgId);
+    cJSON_AddItemToObject(pRoot, "destId", pDestId);
+
+    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->term);
+    cJSON_AddStringToObject(pRoot, "term", u64buf);
+    cJSON_AddNumberToObject(pRoot, "vote_granted", pMsg->voteGranted);
+  }
 
   cJSON* pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SyncRequestVoteReply", pRoot);
@@ -1072,62 +1084,64 @@ SyncAppendEntries* syncAppendEntriesFromRpcMsg2(const SRpcMsg* pRpcMsg) {
 }
 
 cJSON* syncAppendEntries2Json(const SyncAppendEntries* pMsg) {
-  char u64buf[128];
-
+  char   u64buf[128];
   cJSON* pRoot = cJSON_CreateObject();
-  cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
-  cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
-
-  cJSON* pSrcId = cJSON_CreateObject();
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pMsg->srcId.addr);
-  cJSON_AddStringToObject(pSrcId, "addr", u64buf);
-  {
-    uint64_t u64 = pMsg->srcId.addr;
-    cJSON*   pTmp = pSrcId;
-    char     host[128];
-    uint16_t port;
-    syncUtilU642Addr(u64, host, sizeof(host), &port);
-    cJSON_AddStringToObject(pTmp, "addr_host", host);
-    cJSON_AddNumberToObject(pTmp, "addr_port", port);
-  }
-  cJSON_AddNumberToObject(pSrcId, "vgId", pMsg->srcId.vgId);
-  cJSON_AddItemToObject(pRoot, "srcId", pSrcId);
-
-  cJSON* pDestId = cJSON_CreateObject();
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pMsg->destId.addr);
-  cJSON_AddStringToObject(pDestId, "addr", u64buf);
-  {
-    uint64_t u64 = pMsg->destId.addr;
-    cJSON*   pTmp = pDestId;
-    char     host[128];
-    uint16_t port;
-    syncUtilU642Addr(u64, host, sizeof(host), &port);
-    cJSON_AddStringToObject(pTmp, "addr_host", host);
-    cJSON_AddNumberToObject(pTmp, "addr_port", port);
-  }
-  cJSON_AddNumberToObject(pDestId, "vgId", pMsg->destId.vgId);
-  cJSON_AddItemToObject(pRoot, "destId", pDestId);
 
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pMsg->term);
-  cJSON_AddStringToObject(pRoot, "term", u64buf);
-
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pMsg->prevLogIndex);
-  cJSON_AddStringToObject(pRoot, "pre_log_index", u64buf);
-
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pMsg->prevLogTerm);
-  cJSON_AddStringToObject(pRoot, "pre_log_term", u64buf);
-
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pMsg->commitIndex);
-  cJSON_AddStringToObject(pRoot, "commit_index", u64buf);
-
-  cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen);
-  char* s;
-  s = syncUtilprintBin((char*)(pMsg->data), pMsg->dataLen);
-  cJSON_AddStringToObject(pRoot, "data", s);
-  free(s);
-  s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
-  cJSON_AddStringToObject(pRoot, "data2", s);
-  free(s);
+  if (pMsg != NULL) {
+    cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
+    cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
+
+    cJSON* pSrcId = cJSON_CreateObject();
+    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->srcId.addr);
+    cJSON_AddStringToObject(pSrcId, "addr", u64buf);
+    {
+      uint64_t u64 = pMsg->srcId.addr;
+      cJSON*   pTmp = pSrcId;
+      char     host[128];
+      uint16_t port;
+      syncUtilU642Addr(u64, host, sizeof(host), &port);
+      cJSON_AddStringToObject(pTmp, "addr_host", host);
+      cJSON_AddNumberToObject(pTmp, "addr_port", port);
+    }
+    cJSON_AddNumberToObject(pSrcId, "vgId", pMsg->srcId.vgId);
+    cJSON_AddItemToObject(pRoot, "srcId", pSrcId);
+
+    cJSON* pDestId = cJSON_CreateObject();
+    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->destId.addr);
+    cJSON_AddStringToObject(pDestId, "addr", u64buf);
+    {
+      uint64_t u64 = pMsg->destId.addr;
+      cJSON*   pTmp = pDestId;
+      char     host[128];
+      uint16_t port;
+      syncUtilU642Addr(u64, host, sizeof(host), &port);
+      cJSON_AddStringToObject(pTmp, "addr_host", host);
+      cJSON_AddNumberToObject(pTmp, "addr_port", port);
+    }
+    cJSON_AddNumberToObject(pDestId, "vgId", pMsg->destId.vgId);
+    cJSON_AddItemToObject(pRoot, "destId", pDestId);
+
+    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->term);
+    cJSON_AddStringToObject(pRoot, "term", u64buf);
+
+    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->prevLogIndex);
+    cJSON_AddStringToObject(pRoot, "pre_log_index", u64buf);
+
+    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->prevLogTerm);
+    cJSON_AddStringToObject(pRoot, "pre_log_term", u64buf);
+
+    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->commitIndex);
+    cJSON_AddStringToObject(pRoot, "commit_index", u64buf);
+
+    cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen);
+    char* s;
+    s = syncUtilprintBin((char*)(pMsg->data), pMsg->dataLen);
+    cJSON_AddStringToObject(pRoot, "data", s);
+    free(s);
+    s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
+    cJSON_AddStringToObject(pRoot, "data2", s);
+    free(s);
+  }
 
   cJSON* pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SyncAppendEntries", pRoot);
@@ -1231,47 +1245,49 @@ SyncAppendEntriesReply* syncAppendEntriesReplyFromRpcMsg2(const SRpcMsg* pRpcMsg
 }
 
 cJSON* syncAppendEntriesReply2Json(const SyncAppendEntriesReply* pMsg) {
-  char u64buf[128];
-
+  char   u64buf[128];
   cJSON* pRoot = cJSON_CreateObject();
-  cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
-  cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
-
-  cJSON* pSrcId = cJSON_CreateObject();
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pMsg->srcId.addr);
-  cJSON_AddStringToObject(pSrcId, "addr", u64buf);
-  {
-    uint64_t u64 = pMsg->srcId.addr;
-    cJSON*   pTmp = pSrcId;
-    char     host[128];
-    uint16_t port;
-    syncUtilU642Addr(u64, host, sizeof(host), &port);
-    cJSON_AddStringToObject(pTmp, "addr_host", host);
-    cJSON_AddNumberToObject(pTmp, "addr_port", port);
-  }
-  cJSON_AddNumberToObject(pSrcId, "vgId", pMsg->srcId.vgId);
-  cJSON_AddItemToObject(pRoot, "srcId", pSrcId);
-
-  cJSON* pDestId = cJSON_CreateObject();
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pMsg->destId.addr);
-  cJSON_AddStringToObject(pDestId, "addr", u64buf);
-  {
-    uint64_t u64 = pMsg->destId.addr;
-    cJSON*   pTmp = pDestId;
-    char     host[128];
-    uint16_t port;
-    syncUtilU642Addr(u64, host, sizeof(host), &port);
-    cJSON_AddStringToObject(pTmp, "addr_host", host);
-    cJSON_AddNumberToObject(pTmp, "addr_port", port);
+
+  if (pMsg != NULL) {
+    cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
+    cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
+
+    cJSON* pSrcId = cJSON_CreateObject();
+    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->srcId.addr);
+    cJSON_AddStringToObject(pSrcId, "addr", u64buf);
+    {
+      uint64_t u64 = pMsg->srcId.addr;
+      cJSON*   pTmp = pSrcId;
+      char     host[128];
+      uint16_t port;
+      syncUtilU642Addr(u64, host, sizeof(host), &port);
+      cJSON_AddStringToObject(pTmp, "addr_host", host);
+      cJSON_AddNumberToObject(pTmp, "addr_port", port);
+    }
+    cJSON_AddNumberToObject(pSrcId, "vgId", pMsg->srcId.vgId);
+    cJSON_AddItemToObject(pRoot, "srcId", pSrcId);
+
+    cJSON* pDestId = cJSON_CreateObject();
+    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->destId.addr);
+    cJSON_AddStringToObject(pDestId, "addr", u64buf);
+    {
+      uint64_t u64 = pMsg->destId.addr;
+      cJSON*   pTmp = pDestId;
+      char     host[128];
+      uint16_t port;
+      syncUtilU642Addr(u64, host, sizeof(host), &port);
+      cJSON_AddStringToObject(pTmp, "addr_host", host);
+      cJSON_AddNumberToObject(pTmp, "addr_port", port);
+    }
+    cJSON_AddNumberToObject(pDestId, "vgId", pMsg->destId.vgId);
+    cJSON_AddItemToObject(pRoot, "destId", pDestId);
+
+    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->term);
+    cJSON_AddStringToObject(pRoot, "term", u64buf);
+    cJSON_AddNumberToObject(pRoot, "success", pMsg->success);
+    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->matchIndex);
+    cJSON_AddStringToObject(pRoot, "matchIndex", u64buf);
   }
-  cJSON_AddNumberToObject(pDestId, "vgId", pMsg->destId.vgId);
-  cJSON_AddItemToObject(pRoot, "destId", pDestId);
-
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pMsg->term);
-  cJSON_AddStringToObject(pRoot, "term", u64buf);
-  cJSON_AddNumberToObject(pRoot, "success", pMsg->success);
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pMsg->matchIndex);
-  cJSON_AddStringToObject(pRoot, "matchIndex", u64buf);
 
   cJSON* pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SyncAppendEntriesReply", pRoot);

source/libs/sync/src/syncRaftEntry.c

@@ -68,29 +68,31 @@ SSyncRaftEntry* syncEntryDeserialize(const char* buf, uint32_t len) {
 }
 
 cJSON* syncEntry2Json(const SSyncRaftEntry* pEntry) {
-  char u64buf[128];
-
+  char   u64buf[128];
   cJSON* pRoot = cJSON_CreateObject();
-  cJSON_AddNumberToObject(pRoot, "bytes", pEntry->bytes);
-  cJSON_AddNumberToObject(pRoot, "msgType", pEntry->msgType);
-  cJSON_AddNumberToObject(pRoot, "originalRpcType", pEntry->originalRpcType);
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pEntry->seqNum);
-  cJSON_AddStringToObject(pRoot, "seqNum", u64buf);
-  cJSON_AddNumberToObject(pRoot, "isWeak", pEntry->isWeak);
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pEntry->term);
-  cJSON_AddStringToObject(pRoot, "term", u64buf);
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pEntry->index);
-  cJSON_AddStringToObject(pRoot, "index", u64buf);
-  cJSON_AddNumberToObject(pRoot, "dataLen", pEntry->dataLen);
-
-  char* s;
-  s = syncUtilprintBin((char*)(pEntry->data), pEntry->dataLen);
-  cJSON_AddStringToObject(pRoot, "data", s);
-  free(s);
-
-  s = syncUtilprintBin2((char*)(pEntry->data), pEntry->dataLen);
-  cJSON_AddStringToObject(pRoot, "data2", s);
-  free(s);
+
+  if (pEntry != NULL) {
+    cJSON_AddNumberToObject(pRoot, "bytes", pEntry->bytes);
+    cJSON_AddNumberToObject(pRoot, "msgType", pEntry->msgType);
+    cJSON_AddNumberToObject(pRoot, "originalRpcType", pEntry->originalRpcType);
+    snprintf(u64buf, sizeof(u64buf), "%lu", pEntry->seqNum);
+    cJSON_AddStringToObject(pRoot, "seqNum", u64buf);
+    cJSON_AddNumberToObject(pRoot, "isWeak", pEntry->isWeak);
+    snprintf(u64buf, sizeof(u64buf), "%lu", pEntry->term);
+    cJSON_AddStringToObject(pRoot, "term", u64buf);
+    snprintf(u64buf, sizeof(u64buf), "%lu", pEntry->index);
+    cJSON_AddStringToObject(pRoot, "index", u64buf);
+    cJSON_AddNumberToObject(pRoot, "dataLen", pEntry->dataLen);
+
+    char* s;
+    s = syncUtilprintBin((char*)(pEntry->data), pEntry->dataLen);
+    cJSON_AddStringToObject(pRoot, "data", s);
+    free(s);
+
+    s = syncUtilprintBin2((char*)(pEntry->data), pEntry->dataLen);
+    cJSON_AddStringToObject(pRoot, "data2", s);
+    free(s);
+  }
 
   cJSON* pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SSyncRaftEntry", pRoot);

source/libs/sync/src/syncRaftLog.c

@@ -34,7 +34,7 @@ SSyncLogStore* logStoreCreate(SSyncNode* pSyncNode) {
   pLogStore->getLastTerm = logStoreLastTerm;
   pLogStore->updateCommitIndex = logStoreUpdateCommitIndex;
   pLogStore->getCommitIndex = logStoreGetCommitIndex;
-  return pLogStore; // to avoid compiler error
+  return pLogStore;  // to avoid compiler error
 }
 
 void logStoreDestory(SSyncLogStore* pLogStore) {
@@ -59,21 +59,24 @@ int32_t logStoreAppendEntry(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry) {
 
   walFsync(pWal, true);
   free(serialized);
-  return code; // to avoid compiler error
+  return code;  // to avoid compiler error
 }
 
 SSyncRaftEntry* logStoreGetEntry(SSyncLogStore* pLogStore, SyncIndex index) {
   SSyncLogStoreData* pData = pLogStore->data;
   SWal*              pWal = pData->pWal;
-  SSyncRaftEntry*    pEntry;
+  SSyncRaftEntry*    pEntry = NULL;
 
-  SWalReadHandle* pWalHandle = walOpenReadHandle(pWal);
-  walReadWithHandle(pWalHandle, index);
-  pEntry = syncEntryDeserialize(pWalHandle->pHead->head.body, pWalHandle->pHead->head.len);
-  assert(pEntry != NULL);
+  if (index >= SYNC_INDEX_BEGIN && index <= logStoreLastIndex(pLogStore)) {
+    SWalReadHandle* pWalHandle = walOpenReadHandle(pWal);
+    walReadWithHandle(pWalHandle, index);
+    pEntry = syncEntryDeserialize(pWalHandle->pHead->head.body, pWalHandle->pHead->head.len);
+    assert(pEntry != NULL);
+
+    // need to hold, do not new every time!!
+    walCloseReadHandle(pWalHandle);
+  }
 
-  // need to hold, do not new every time!!
-  walCloseReadHandle(pWalHandle);
   return pEntry;
 }
 
@@ -81,7 +84,7 @@ int32_t logStoreTruncate(SSyncLogStore* pLogStore, SyncIndex fromIndex) {
   SSyncLogStoreData* pData = pLogStore->data;
   SWal*              pWal = pData->pWal;
   walRollback(pWal, fromIndex);
-  return 0; // to avoid compiler error
+  return 0;  // to avoid compiler error
 }
 
 SyncIndex logStoreLastIndex(SSyncLogStore* pLogStore) {
@@ -105,7 +108,7 @@ int32_t logStoreUpdateCommitIndex(SSyncLogStore* pLogStore, SyncIndex index) {
   SSyncLogStoreData* pData = pLogStore->data;
   SWal*              pWal = pData->pWal;
   walCommit(pWal, index);
-  return 0; // to avoid compiler error
+  return 0;  // to avoid compiler error
 }
 
 SyncIndex logStoreGetCommitIndex(SSyncLogStore* pLogStore) {
@@ -126,26 +129,28 @@ SSyncRaftEntry* logStoreGetLastEntry(SSyncLogStore* pLogStore) {
 }
 
 cJSON* logStore2Json(SSyncLogStore* pLogStore) {
-  char u64buf[128];
-
+  char               u64buf[128];
   SSyncLogStoreData* pData = (SSyncLogStoreData*)pLogStore->data;
   cJSON*             pRoot = cJSON_CreateObject();
-  snprintf(u64buf, sizeof(u64buf), "%p", pData->pSyncNode);
-  cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf);
-  snprintf(u64buf, sizeof(u64buf), "%p", pData->pWal);
-  cJSON_AddStringToObject(pRoot, "pWal", u64buf);
-  snprintf(u64buf, sizeof(u64buf), "%" PRId64 "", logStoreLastIndex(pLogStore));
-  cJSON_AddStringToObject(pRoot, "LastIndex", u64buf);
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", logStoreLastTerm(pLogStore));
-  cJSON_AddStringToObject(pRoot, "LastTerm", u64buf);
-
-  cJSON* pEntries = cJSON_CreateArray();
-  cJSON_AddItemToObject(pRoot, "pEntries", pEntries);
-  SyncIndex lastIndex = logStoreLastIndex(pLogStore);
-  for (SyncIndex i = 0; i <= lastIndex; ++i) {
-    SSyncRaftEntry* pEntry = logStoreGetEntry(pLogStore, i);
-    cJSON_AddItemToArray(pEntries, syncEntry2Json(pEntry));
-    syncEntryDestory(pEntry);
+
+  if (pData != NULL && pData->pWal != NULL) {
+    snprintf(u64buf, sizeof(u64buf), "%p", pData->pSyncNode);
+    cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf);
+    snprintf(u64buf, sizeof(u64buf), "%p", pData->pWal);
+    cJSON_AddStringToObject(pRoot, "pWal", u64buf);
+    snprintf(u64buf, sizeof(u64buf), "%ld", logStoreLastIndex(pLogStore));
+    cJSON_AddStringToObject(pRoot, "LastIndex", u64buf);
+    snprintf(u64buf, sizeof(u64buf), "%lu", logStoreLastTerm(pLogStore));
+    cJSON_AddStringToObject(pRoot, "LastTerm", u64buf);
+
+    cJSON* pEntries = cJSON_CreateArray();
+    cJSON_AddItemToObject(pRoot, "pEntries", pEntries);
+    SyncIndex lastIndex = logStoreLastIndex(pLogStore);
+    for (SyncIndex i = 0; i <= lastIndex; ++i) {
+      SSyncRaftEntry* pEntry = logStoreGetEntry(pLogStore, i);
+      cJSON_AddItemToArray(pEntries, syncEntry2Json(pEntry));
+      syncEntryDestory(pEntry);
+    }
   }
 
   cJSON* pJson = cJSON_CreateObject();

source/libs/sync/src/syncRaftStore.c

@@ -97,16 +97,32 @@ int32_t raftStorePersist(SRaftStore *pRaftStore) {
   return 0;
 }
 
-static bool raftStoreFileExist(char *path) { return taosStatFile(path, NULL, NULL) >= 0; }
+static bool raftStoreFileExist(char *path) {
+  bool b = taosStatFile(path, NULL, NULL) >= 0;
+  return b;
+}
 
 int32_t raftStoreSerialize(SRaftStore *pRaftStore, char *buf, size_t len) {
   assert(pRaftStore != NULL);
 
   cJSON *pRoot = cJSON_CreateObject();
-  cJSON_AddNumberToObject(pRoot, "current_term", pRaftStore->currentTerm);
-  cJSON_AddNumberToObject(pRoot, "vote_for_addr", pRaftStore->voteFor.addr);
+
+  char u64Buf[128];
+  snprintf(u64Buf, sizeof(u64Buf), "%lu", pRaftStore->currentTerm);
+  cJSON_AddStringToObject(pRoot, "current_term", u64Buf);
+
+  snprintf(u64Buf, sizeof(u64Buf), "%lu", pRaftStore->voteFor.addr);
+  cJSON_AddStringToObject(pRoot, "vote_for_addr", u64Buf);
+
   cJSON_AddNumberToObject(pRoot, "vote_for_vgid", pRaftStore->voteFor.vgId);
 
+  uint64_t u64 = pRaftStore->voteFor.addr;
+  char     host[128];
+  uint16_t port;
+  syncUtilU642Addr(u64, host, sizeof(host), &port);
+  cJSON_AddStringToObject(pRoot, "addr_host", host);
+  cJSON_AddNumberToObject(pRoot, "addr_port", port);
+
   char *serialized = cJSON_Print(pRoot);
   int   len2 = strlen(serialized);
   assert(len2 < len);
@@ -125,10 +141,12 @@ int32_t raftStoreDeserialize(SRaftStore *pRaftStore, char *buf, size_t len) {
   cJSON *pRoot = cJSON_Parse(buf);
 
   cJSON *pCurrentTerm = cJSON_GetObjectItem(pRoot, "current_term");
-  pRaftStore->currentTerm = pCurrentTerm->valueint;
+  assert(cJSON_IsString(pCurrentTerm));
+  sscanf(pCurrentTerm->valuestring, "%lu", &(pRaftStore->currentTerm));
 
   cJSON *pVoteForAddr = cJSON_GetObjectItem(pRoot, "vote_for_addr");
-  pRaftStore->voteFor.addr = pVoteForAddr->valueint;
+  assert(cJSON_IsString(pVoteForAddr));
+  sscanf(pVoteForAddr->valuestring, "%lu", &(pRaftStore->voteFor.addr));
 
   cJSON *pVoteForVgid = cJSON_GetObjectItem(pRoot, "vote_for_vgid");
   pRaftStore->voteFor.vgId = pVoteForVgid->valueint;
@@ -139,11 +157,10 @@ int32_t raftStoreDeserialize(SRaftStore *pRaftStore, char *buf, size_t len) {
 
 bool raftStoreHasVoted(SRaftStore *pRaftStore) {
   bool b = syncUtilEmptyId(&(pRaftStore->voteFor));
-  return b;
+  return (!b);
 }
 
 void raftStoreVote(SRaftStore *pRaftStore, SRaftId *pRaftId) {
-  assert(!raftStoreHasVoted(pRaftStore));
   assert(!syncUtilEmptyId(pRaftId));
   pRaftStore->voteFor = *pRaftId;
   raftStorePersist(pRaftStore);
@@ -164,30 +181,68 @@ void raftStoreSetTerm(SRaftStore *pRaftStore, SyncTerm term) {
   raftStorePersist(pRaftStore);
 }
 
+int32_t raftStoreFromJson(SRaftStore *pRaftStore, cJSON *pJson) { return 0; }
+
+cJSON *raftStore2Json(SRaftStore *pRaftStore) {
+  char   u64buf[128];
+  cJSON *pRoot = cJSON_CreateObject();
+
+  if (pRaftStore != NULL) {
+    snprintf(u64buf, sizeof(u64buf), "%lu", pRaftStore->currentTerm);
+    cJSON_AddStringToObject(pRoot, "currentTerm", u64buf);
+
+    cJSON *pVoteFor = cJSON_CreateObject();
+    snprintf(u64buf, sizeof(u64buf), "%lu", pRaftStore->voteFor.addr);
+    cJSON_AddStringToObject(pVoteFor, "addr", u64buf);
+    {
+      uint64_t u64 = pRaftStore->voteFor.addr;
+      char     host[128];
+      uint16_t port;
+      syncUtilU642Addr(u64, host, sizeof(host), &port);
+      cJSON_AddStringToObject(pVoteFor, "addr_host", host);
+      cJSON_AddNumberToObject(pVoteFor, "addr_port", port);
+    }
+    cJSON_AddNumberToObject(pVoteFor, "vgId", pRaftStore->voteFor.vgId);
+    cJSON_AddItemToObject(pRoot, "voteFor", pVoteFor);
+
+    int hasVoted = raftStoreHasVoted(pRaftStore);
+    cJSON_AddNumberToObject(pRoot, "hasVoted", hasVoted);
+  }
+
+  cJSON *pJson = cJSON_CreateObject();
+  cJSON_AddItemToObject(pJson, "SRaftStore", pRoot);
+  return pJson;
+}
+
+char *raftStore2Str(SRaftStore *pRaftStore) {
+  cJSON *pJson = raftStore2Json(pRaftStore);
+  char * serialized = cJSON_Print(pJson);
+  cJSON_Delete(pJson);
+  return serialized;
+}
+
 // for debug -------------------
 void raftStorePrint(SRaftStore *pObj) {
-  char serialized[RAFT_STORE_BLOCK_SIZE];
-  raftStoreSerialize(pObj, serialized, sizeof(serialized));
+  char *serialized = raftStore2Str(pObj);
   printf("raftStorePrint | len:%lu | %s \n", strlen(serialized), serialized);
   fflush(NULL);
+  free(serialized);
 }
 
 void raftStorePrint2(char *s, SRaftStore *pObj) {
-  char serialized[RAFT_STORE_BLOCK_SIZE];
-  raftStoreSerialize(pObj, serialized, sizeof(serialized));
+  char *serialized = raftStore2Str(pObj);
   printf("raftStorePrint2 | len:%lu | %s | %s \n", strlen(serialized), s, serialized);
   fflush(NULL);
+  free(serialized);
 }
 void raftStoreLog(SRaftStore *pObj) {
-  char serialized[RAFT_STORE_BLOCK_SIZE];
-  raftStoreSerialize(pObj, serialized, sizeof(serialized));
+  char *serialized = raftStore2Str(pObj);
   sTrace("raftStoreLog | len:%lu | %s", strlen(serialized), serialized);
-  fflush(NULL);
+  free(serialized);
 }
 
 void raftStoreLog2(char *s, SRaftStore *pObj) {
-  char serialized[RAFT_STORE_BLOCK_SIZE];
-  raftStoreSerialize(pObj, serialized, sizeof(serialized));
+  char *serialized = raftStore2Str(pObj);
   sTrace("raftStoreLog2 | len:%lu | %s | %s", strlen(serialized), s, serialized);
-  fflush(NULL);
+  free(serialized);
 }

source/libs/sync/src/syncReplication.c

@@ -18,6 +18,7 @@
 #include "syncMessage.h"
 #include "syncRaftEntry.h"
 #include "syncRaftLog.h"
+#include "syncRaftStore.h"
 #include "syncUtil.h"
 
 // TLA+ Spec
@@ -50,33 +51,54 @@ int32_t syncNodeAppendEntriesPeers(SSyncNode* pSyncNode) {
 
   int32_t ret = 0;
   for (int i = 0; i < pSyncNode->peersNum; ++i) {
-    SRaftId*  pDestId = &(pSyncNode->peersId[i]);
-    SyncIndex nextIndex = syncIndexMgrGetIndex(pSyncNode->pNextIndex, pDestId);
+    SRaftId* pDestId = &(pSyncNode->peersId[i]);
 
+    // set prevLogIndex
+    SyncIndex nextIndex = syncIndexMgrGetIndex(pSyncNode->pNextIndex, pDestId);
     SyncIndex preLogIndex = nextIndex - 1;
 
+    // set preLogTerm
     SyncTerm preLogTerm = 0;
     if (preLogIndex >= SYNC_INDEX_BEGIN) {
       SSyncRaftEntry* pPreEntry = pSyncNode->pLogStore->getEntry(pSyncNode->pLogStore, preLogIndex);
+      assert(pPreEntry != NULL);
+
       preLogTerm = pPreEntry->term;
+      syncEntryDestory(pPreEntry);
     }
 
-    SyncIndex lastIndex = syncUtilMinIndex(pSyncNode->pLogStore->getLastIndex(pSyncNode->pLogStore), nextIndex);
-    assert(nextIndex == lastIndex);
+    // batch optimized
+    // SyncIndex lastIndex = syncUtilMinIndex(pSyncNode->pLogStore->getLastIndex(pSyncNode->pLogStore), nextIndex);
+
+    SyncAppendEntries* pMsg = NULL;
+    SSyncRaftEntry*    pEntry = logStoreGetEntry(pSyncNode->pLogStore, nextIndex);
+    if (pEntry != NULL) {
+      SyncAppendEntries* pMsg = syncAppendEntriesBuild(pEntry->bytes);
 
-    SSyncRaftEntry* pEntry = logStoreGetEntry(pSyncNode->pLogStore, nextIndex);
-    assert(pEntry != NULL);
+      // add pEntry into msg
+      uint32_t len;
+      char*    serialized = syncEntrySerialize(pEntry, &len);
+      assert(len == pEntry->bytes);
+      memcpy(pMsg->data, serialized, len);
+
+      free(serialized);
+      syncEntryDestory(pEntry);
+
+    } else {
+      // maybe overflow, send empty record
+      SyncAppendEntries* pMsg = syncAppendEntriesBuild(0);
+    }
 
-    SyncAppendEntries* pMsg = syncAppendEntriesBuild(pEntry->bytes);
     pMsg->srcId = pSyncNode->myRaftId;
     pMsg->destId = *pDestId;
+    pMsg->term = pSyncNode->pRaftStore->currentTerm;
     pMsg->prevLogIndex = preLogIndex;
     pMsg->prevLogTerm = preLogTerm;
     pMsg->commitIndex = pSyncNode->commitIndex;
-    pMsg->dataLen = pEntry->bytes;
-    // add pEntry into msg
 
+    // send AppendEntries
     syncNodeAppendEntries(pSyncNode, pDestId, pMsg);
+    syncAppendEntriesDestroy(pMsg);
   }
 
   return ret;

source/libs/sync/src/syncRequestVote.c

@@ -56,6 +56,8 @@ int32_t syncNodeOnRequestVoteCb(SSyncNode* ths, SyncRequestVote* pMsg) {
   bool grant = (pMsg->term == ths->pRaftStore->currentTerm) && logOK &&
                ((!raftStoreHasVoted(ths->pRaftStore)) || (syncUtilSameId(&(ths->pRaftStore->voteFor), &(pMsg->srcId))));
   if (grant) {
+    // maybe has already voted for pMsg->srcId
+    // vote again, no harm
     raftStoreVote(ths->pRaftStore, &(pMsg->srcId));
   }
 

source/libs/sync/src/syncRequestVoteReply.c

@@ -45,6 +45,7 @@ int32_t syncNodeOnRequestVoteReplyCb(SSyncNode* ths, SyncRequestVoteReply* pMsg)
     return ret;
   }
 
+  assert(!(pMsg->term > ths->pRaftStore->currentTerm));
   // no need this code, because if I receive reply.term, then I must have sent for that term.
   //  if (pMsg->term > ths->pRaftStore->currentTerm) {
   //    syncNodeUpdateTerm(ths, pMsg->term);
@@ -52,17 +53,29 @@ int32_t syncNodeOnRequestVoteReplyCb(SSyncNode* ths, SyncRequestVoteReply* pMsg)
 
   assert(pMsg->term == ths->pRaftStore->currentTerm);
 
+  // This tallies votes even when the current state is not Candidate,
+  // but they won't be looked at, so it doesn't matter.
   if (ths->state == TAOS_SYNC_STATE_CANDIDATE) {
     votesRespondAdd(ths->pVotesRespond, pMsg);
     if (pMsg->voteGranted) {
+      // add vote
       voteGrantedVote(ths->pVotesGranted, pMsg);
+
+      // maybe to leader
       if (voteGrantedMajority(ths->pVotesGranted)) {
-        if (ths->pVotesGranted->toLeader) {
+        if (!ths->pVotesGranted->toLeader) {
           syncNodeCandidate2Leader(ths);
+
+          // prevent to leader again!
           ths->pVotesGranted->toLeader = true;
         }
       }
+    } else {
+      ;
+      // do nothing
+      // UNCHANGED <<votesGranted, voterLog>>
     }
   }
+
   return ret;
 }

source/libs/sync/src/syncTimeout.c

@@ -24,7 +24,8 @@ int32_t syncNodeOnTimeoutCb(SSyncNode* ths, SyncTimeout* pMsg) {
   if (pMsg->timeoutType == SYNC_TIMEOUT_PING) {
     if (atomic_load_64(&ths->pingTimerLogicClockUser) <= pMsg->logicClock) {
       ++(ths->pingTimerCounter);
-      syncNodePingAll(ths);
+      // syncNodePingAll(ths);
+      syncNodePingPeers(ths);
     }
 
   } else if (pMsg->timeoutType == SYNC_TIMEOUT_ELECTION) {

source/libs/sync/src/syncVoteMgr.c

@@ -82,30 +82,32 @@ cJSON *voteGranted2Json(SVotesGranted *pVotesGranted) {
   char   u64buf[128];
   cJSON *pRoot = cJSON_CreateObject();
 
-  cJSON_AddNumberToObject(pRoot, "replicaNum", pVotesGranted->replicaNum);
-  cJSON *pReplicas = cJSON_CreateArray();
-  cJSON_AddItemToObject(pRoot, "replicas", pReplicas);
-  for (int i = 0; i < pVotesGranted->replicaNum; ++i) {
-    cJSON_AddItemToArray(pReplicas, syncUtilRaftId2Json(&(*(pVotesGranted->replicas))[i]));
-  }
-  int *arr = (int *)malloc(sizeof(int) * pVotesGranted->replicaNum);
-  for (int i = 0; i < pVotesGranted->replicaNum; ++i) {
-    arr[i] = pVotesGranted->isGranted[i];
+  if (pVotesGranted != NULL) {
+    cJSON_AddNumberToObject(pRoot, "replicaNum", pVotesGranted->replicaNum);
+    cJSON *pReplicas = cJSON_CreateArray();
+    cJSON_AddItemToObject(pRoot, "replicas", pReplicas);
+    for (int i = 0; i < pVotesGranted->replicaNum; ++i) {
+      cJSON_AddItemToArray(pReplicas, syncUtilRaftId2Json(&(*(pVotesGranted->replicas))[i]));
+    }
+    int *arr = (int *)malloc(sizeof(int) * pVotesGranted->replicaNum);
+    for (int i = 0; i < pVotesGranted->replicaNum; ++i) {
+      arr[i] = pVotesGranted->isGranted[i];
+    }
+    cJSON *pIsGranted = cJSON_CreateIntArray(arr, pVotesGranted->replicaNum);
+    free(arr);
+    cJSON_AddItemToObject(pRoot, "isGranted", pIsGranted);
+
+    cJSON_AddNumberToObject(pRoot, "votes", pVotesGranted->votes);
+    snprintf(u64buf, sizeof(u64buf), "%lu", pVotesGranted->term);
+    cJSON_AddStringToObject(pRoot, "term", u64buf);
+    cJSON_AddNumberToObject(pRoot, "quorum", pVotesGranted->quorum);
+    cJSON_AddNumberToObject(pRoot, "toLeader", pVotesGranted->toLeader);
+    snprintf(u64buf, sizeof(u64buf), "%p", pVotesGranted->pSyncNode);
+    cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf);
+
+    bool majority = voteGrantedMajority(pVotesGranted);
+    cJSON_AddNumberToObject(pRoot, "majority", majority);
   }
-  cJSON *pIsGranted = cJSON_CreateIntArray(arr, pVotesGranted->replicaNum);
-  free(arr);
-  cJSON_AddItemToObject(pRoot, "isGranted", pIsGranted);
-
-  cJSON_AddNumberToObject(pRoot, "votes", pVotesGranted->votes);
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pVotesGranted->term);
-  cJSON_AddStringToObject(pRoot, "term", u64buf);
-  cJSON_AddNumberToObject(pRoot, "quorum", pVotesGranted->quorum);
-  cJSON_AddNumberToObject(pRoot, "toLeader", pVotesGranted->toLeader);
-  snprintf(u64buf, sizeof(u64buf), "%p", pVotesGranted->pSyncNode);
-  cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf);
-
-  bool majority = voteGrantedMajority(pVotesGranted);
-  cJSON_AddNumberToObject(pRoot, "majority", majority);
 
   cJSON *pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SVotesGranted", pRoot);
@@ -122,27 +124,27 @@ char *voteGranted2Str(SVotesGranted *pVotesGranted) {
 // for debug -------------------
 void voteGrantedPrint(SVotesGranted *pObj) {
   char *serialized = voteGranted2Str(pObj);
-  printf("voteGrantedPrint | len:%zu | %s \n", strlen(serialized), serialized);
+  printf("voteGrantedPrint | len:%lu | %s \n", strlen(serialized), serialized);
   fflush(NULL);
   free(serialized);
 }
 
 void voteGrantedPrint2(char *s, SVotesGranted *pObj) {
   char *serialized = voteGranted2Str(pObj);
-  printf("voteGrantedPrint2 | len:%zu | %s | %s \n", strlen(serialized), s, serialized);
+  printf("voteGrantedPrint2 | len:%lu | %s | %s \n", strlen(serialized), s, serialized);
   fflush(NULL);
   free(serialized);
 }
 
 void voteGrantedLog(SVotesGranted *pObj) {
   char *serialized = voteGranted2Str(pObj);
-  sTrace("voteGrantedLog | len:%zu | %s", strlen(serialized), serialized);
+  sTrace("voteGrantedLog | len:%lu | %s", strlen(serialized), serialized);
   free(serialized);
 }
 
 void voteGrantedLog2(char *s, SVotesGranted *pObj) {
   char *serialized = voteGranted2Str(pObj);
-  sTrace("voteGrantedLog2 | len:%zu | %s | %s", strlen(serialized), s, serialized);
+  sTrace("voteGrantedLog2 | len:%lu | %s | %s", strlen(serialized), s, serialized);
   free(serialized);
 }
 
@@ -203,29 +205,31 @@ cJSON *votesRespond2Json(SVotesRespond *pVotesRespond) {
   char   u64buf[128];
   cJSON *pRoot = cJSON_CreateObject();
 
-  cJSON_AddNumberToObject(pRoot, "replicaNum", pVotesRespond->replicaNum);
-  cJSON *pReplicas = cJSON_CreateArray();
-  cJSON_AddItemToObject(pRoot, "replicas", pReplicas);
-  for (int i = 0; i < pVotesRespond->replicaNum; ++i) {
-    cJSON_AddItemToArray(pReplicas, syncUtilRaftId2Json(&(*(pVotesRespond->replicas))[i]));
-  }
-  int  respondNum = 0;
-  int *arr = (int *)malloc(sizeof(int) * pVotesRespond->replicaNum);
-  for (int i = 0; i < pVotesRespond->replicaNum; ++i) {
-    arr[i] = pVotesRespond->isRespond[i];
-    if (pVotesRespond->isRespond[i]) {
-      respondNum++;
+  if (pVotesRespond != NULL) {
+    cJSON_AddNumberToObject(pRoot, "replicaNum", pVotesRespond->replicaNum);
+    cJSON *pReplicas = cJSON_CreateArray();
+    cJSON_AddItemToObject(pRoot, "replicas", pReplicas);
+    for (int i = 0; i < pVotesRespond->replicaNum; ++i) {
+      cJSON_AddItemToArray(pReplicas, syncUtilRaftId2Json(&(*(pVotesRespond->replicas))[i]));
+    }
+    int  respondNum = 0;
+    int *arr = (int *)malloc(sizeof(int) * pVotesRespond->replicaNum);
+    for (int i = 0; i < pVotesRespond->replicaNum; ++i) {
+      arr[i] = pVotesRespond->isRespond[i];
+      if (pVotesRespond->isRespond[i]) {
+        respondNum++;
+      }
     }
+    cJSON *pIsRespond = cJSON_CreateIntArray(arr, pVotesRespond->replicaNum);
+    free(arr);
+    cJSON_AddItemToObject(pRoot, "isRespond", pIsRespond);
+    cJSON_AddNumberToObject(pRoot, "respondNum", respondNum);
+
+    snprintf(u64buf, sizeof(u64buf), "%lu", pVotesRespond->term);
+    cJSON_AddStringToObject(pRoot, "term", u64buf);
+    snprintf(u64buf, sizeof(u64buf), "%p", pVotesRespond->pSyncNode);
+    cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf);
   }
-  cJSON *pIsRespond = cJSON_CreateIntArray(arr, pVotesRespond->replicaNum);
-  free(arr);
-  cJSON_AddItemToObject(pRoot, "isRespond", pIsRespond);
-  cJSON_AddNumberToObject(pRoot, "respondNum", respondNum);
-
-  snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pVotesRespond->term);
-  cJSON_AddStringToObject(pRoot, "term", u64buf);
-  snprintf(u64buf, sizeof(u64buf), "%p", pVotesRespond->pSyncNode);
-  cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf);
 
   cJSON *pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SVotesRespond", pRoot);
@@ -242,26 +246,26 @@ char *votesRespond2Str(SVotesRespond *pVotesRespond) {
 // for debug -------------------
 void votesRespondPrint(SVotesRespond *pObj) {
   char *serialized = votesRespond2Str(pObj);
-  printf("votesRespondPrint | len:%zu | %s \n", strlen(serialized), serialized);
+  printf("votesRespondPrint | len:%lu | %s \n", strlen(serialized), serialized);
   fflush(NULL);
   free(serialized);
 }
 
 void votesRespondPrint2(char *s, SVotesRespond *pObj) {
   char *serialized = votesRespond2Str(pObj);
-  printf("votesRespondPrint2 | len:%zu | %s | %s \n", strlen(serialized), s, serialized);
+  printf("votesRespondPrint2 | len:%lu | %s | %s \n", strlen(serialized), s, serialized);
   fflush(NULL);
   free(serialized);
 }
 
 void votesRespondLog(SVotesRespond *pObj) {
   char *serialized = votesRespond2Str(pObj);
-  sTrace("votesRespondLog | len:%zu | %s", strlen(serialized), serialized);
+  sTrace("votesRespondLog | len:%lu | %s", strlen(serialized), serialized);
   free(serialized);
 }
 
 void votesRespondLog2(char *s, SVotesRespond *pObj) {
   char *serialized = votesRespond2Str(pObj);
-  sTrace("votesRespondLog2 | len:%zu | %s | %s", strlen(serialized), s, serialized);
+  sTrace("votesRespondLog2 | len:%lu | %s | %s", strlen(serialized), s, serialized);
   free(serialized);
 }

source/libs/sync/test/CMakeLists.txt

@@ -25,6 +25,9 @@ add_executable(syncTimeoutTest "")
 add_executable(syncPingTest "")
 add_executable(syncPingReplyTest "")
 add_executable(syncRpcMsgTest "")
+add_executable(syncPingTimerTest2 "")
+add_executable(syncPingSelfTest "")
+add_executable(syncElectTest "")
 
 
 target_sources(syncTest
@@ -135,6 +138,18 @@ target_sources(syncRpcMsgTest
     PRIVATE
     "syncRpcMsgTest.cpp"
 )
+target_sources(syncPingTimerTest2
+    PRIVATE
+    "syncPingTimerTest2.cpp"
+)
+target_sources(syncPingSelfTest
+    PRIVATE
+    "syncPingSelfTest.cpp"
+)
+target_sources(syncElectTest
+    PRIVATE
+    "syncElectTest.cpp"
+)
 
 
 target_include_directories(syncTest
@@ -272,6 +287,26 @@ target_include_directories(syncRpcMsgTest
     "${CMAKE_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
+target_include_directories(syncPingTimerTest2
+    PUBLIC
+    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
+)
+target_include_directories(syncPingSelfTest
+    PUBLIC
+    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
+)
+target_include_directories(syncElectTest
+    PUBLIC
+    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
+)
+target_include_directories(syncElectTest
+    PUBLIC
+    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
+)
 
 
 target_link_libraries(syncTest
@@ -382,6 +417,18 @@ target_link_libraries(syncRpcMsgTest
     sync
     gtest_main
 )
+target_link_libraries(syncPingTimerTest2
+    sync
+    gtest_main
+)
+target_link_libraries(syncPingSelfTest
+    sync
+    gtest_main
+)
+target_link_libraries(syncElectTest
+    sync
+    gtest_main
+)
 
 
 enable_testing()

source/libs/sync/test/syncElectTest.cpp

+#include <gtest/gtest.h>
+#include <stdio.h>
+#include "syncEnv.h"
+#include "syncIO.h"
+#include "syncInt.h"
+#include "syncRaftLog.h"
+#include "syncRaftStore.h"
+#include "syncUtil.h"
+
+void logTest() {
+  sTrace("--- sync log test: trace");
+  sDebug("--- sync log test: debug");
+  sInfo("--- sync log test: info");
+  sWarn("--- sync log test: warn");
+  sError("--- sync log test: error");
+  sFatal("--- sync log test: fatal");
+}
+
+uint16_t ports[] = {7010, 7110, 7210, 7310, 7410};
+int32_t  replicaNum = 3;
+int32_t  myIndex = 0;
+
+SRaftId    ids[TSDB_MAX_REPLICA];
+SSyncInfo  syncInfo;
+SSyncFSM*  pFsm;
+SWal*      pWal;
+SSyncNode* gSyncNode;
+
+SSyncNode* syncNodeInit() {
+  syncInfo.vgId = 1234;
+  syncInfo.rpcClient = gSyncIO->clientRpc;
+  syncInfo.FpSendMsg = syncIOSendMsg;
+  syncInfo.queue = gSyncIO->pMsgQ;
+  syncInfo.FpEqMsg = syncIOEqMsg;
+  syncInfo.pFsm = pFsm;
+  snprintf(syncInfo.path, sizeof(syncInfo.path), "./elect_test_%d", myIndex);
+
+  int code = walInit();
+  assert(code == 0);
+  SWalCfg walCfg;
+  memset(&walCfg, 0, sizeof(SWalCfg));
+  walCfg.vgId = syncInfo.vgId;
+  walCfg.fsyncPeriod = 1000;
+  walCfg.retentionPeriod = 1000;
+  walCfg.rollPeriod = 1000;
+  walCfg.retentionSize = 1000;
+  walCfg.segSize = 1000;
+  walCfg.level = TAOS_WAL_FSYNC;
+
+  char tmpdir[128];
+  snprintf(tmpdir, sizeof(tmpdir), "./elect_test_wal_%d", myIndex);
+  pWal = walOpen(tmpdir, &walCfg);
+  assert(pWal != NULL);
+
+  syncInfo.pWal = pWal;
+
+  SSyncCfg* pCfg = &syncInfo.syncCfg;
+  pCfg->myIndex = myIndex;
+  pCfg->replicaNum = replicaNum;
+
+  for (int i = 0; i < replicaNum; ++i) {
+    pCfg->nodeInfo[i].nodePort = ports[i];
+    snprintf(pCfg->nodeInfo[i].nodeFqdn, sizeof(pCfg->nodeInfo[i].nodeFqdn), "%s", "127.0.0.1");
+    // taosGetFqdn(pCfg->nodeInfo[0].nodeFqdn);
+  }
+
+  SSyncNode* pSyncNode = syncNodeOpen(&syncInfo);
+  assert(pSyncNode != NULL);
+
+  gSyncIO->FpOnSyncPing = pSyncNode->FpOnPing;
+  gSyncIO->FpOnSyncPingReply = pSyncNode->FpOnPingReply;
+  gSyncIO->FpOnSyncRequestVote = pSyncNode->FpOnRequestVote;
+  gSyncIO->FpOnSyncRequestVoteReply = pSyncNode->FpOnRequestVoteReply;
+  gSyncIO->FpOnSyncAppendEntries = pSyncNode->FpOnAppendEntries;
+  gSyncIO->FpOnSyncAppendEntriesReply = pSyncNode->FpOnAppendEntriesReply;
+  gSyncIO->FpOnSyncPing = pSyncNode->FpOnPing;
+  gSyncIO->FpOnSyncPingReply = pSyncNode->FpOnPingReply;
+  gSyncIO->FpOnSyncTimeout = pSyncNode->FpOnTimeout;
+  gSyncIO->pSyncNode = pSyncNode;
+
+  return pSyncNode;
+}
+
+SSyncNode* syncInitTest() { return syncNodeInit(); }
+
+void initRaftId(SSyncNode* pSyncNode) {
+  for (int i = 0; i < replicaNum; ++i) {
+    ids[i] = pSyncNode->replicasId[i];
+    char* s = syncUtilRaftId2Str(&ids[i]);
+    printf("raftId[%d] : %s\n", i, s);
+    free(s);
+  }
+}
+
+int main(int argc, char** argv) {
+  // taosInitLog((char *)"syncTest.log", 100000, 10);
+  tsAsyncLog = 0;
+  sDebugFlag = 143 + 64;
+
+  myIndex = 0;
+  if (argc >= 2) {
+    myIndex = atoi(argv[1]);
+  }
+
+  int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]);
+  assert(ret == 0);
+
+  ret = syncEnvStart();
+  assert(ret == 0);
+
+  gSyncNode = syncInitTest();
+  assert(gSyncNode != NULL);
+  syncNodePrint2((char*)"", gSyncNode);
+
+  initRaftId(gSyncNode);
+
+  //---------------------------
+  while (1) {
+    sTrace("while 1 sleep, state: %d, %s", gSyncNode->state, syncUtilState2String(gSyncNode->state));
+    taosMsleep(1000);
+  }
+
+  return 0;
+}

source/libs/sync/test/syncInitTest.cpp

@@ -47,12 +47,11 @@ SSyncNode* syncNodeInit() {
 
   gSyncIO->FpOnSyncPing = pSyncNode->FpOnPing;
   gSyncIO->FpOnSyncPingReply = pSyncNode->FpOnPingReply;
+  gSyncIO->FpOnSyncClientRequest = pSyncNode->FpOnClientRequest;
   gSyncIO->FpOnSyncRequestVote = pSyncNode->FpOnRequestVote;
   gSyncIO->FpOnSyncRequestVoteReply = pSyncNode->FpOnRequestVoteReply;
   gSyncIO->FpOnSyncAppendEntries = pSyncNode->FpOnAppendEntries;
   gSyncIO->FpOnSyncAppendEntriesReply = pSyncNode->FpOnAppendEntriesReply;
-  gSyncIO->FpOnSyncPing = pSyncNode->FpOnPing;
-  gSyncIO->FpOnSyncPingReply = pSyncNode->FpOnPingReply;
   gSyncIO->FpOnSyncTimeout = pSyncNode->FpOnTimeout;
   gSyncIO->pSyncNode = pSyncNode;
 

source/libs/sync/test/syncLogStoreTest.cpp

@@ -81,7 +81,7 @@ SSyncNode* syncNodeInit() {
 SSyncNode* syncInitTest() { return syncNodeInit(); }
 
 void logStoreTest() {
-  logStorePrint2((char*)"logStoreTest2", pSyncNode->pLogStore);
+  logStorePrint2((char*)"logStoreTest", pSyncNode->pLogStore);
 
   assert(pSyncNode->pLogStore->getLastIndex(pSyncNode->pLogStore) == SYNC_INDEX_INVALID);
 
@@ -105,10 +105,10 @@ void logStoreTest() {
       assert(pSyncNode->pLogStore->getLastIndex(pSyncNode->pLogStore) == SYNC_INDEX_BEGIN);
     }
   }
-  logStorePrint(pSyncNode->pLogStore);
+  logStorePrint2((char*)"after appendEntry", pSyncNode->pLogStore);
 
   pSyncNode->pLogStore->truncate(pSyncNode->pLogStore, 3);
-  logStorePrint(pSyncNode->pLogStore);
+  logStorePrint2((char*)"after truncate 3", pSyncNode->pLogStore);
 }
 
 void initRaftId(SSyncNode* pSyncNode) {

source/libs/sync/test/syncPingSelfTest.cpp

+#include <gtest/gtest.h>
+#include <stdio.h>
+#include "syncEnv.h"
+#include "syncIO.h"
+#include "syncInt.h"
+#include "syncRaftStore.h"
+#include "syncUtil.h"
+
+void logTest() {
+  sTrace("--- sync log test: trace");
+  sDebug("--- sync log test: debug");
+  sInfo("--- sync log test: info");
+  sWarn("--- sync log test: warn");
+  sError("--- sync log test: error");
+  sFatal("--- sync log test: fatal");
+}
+
+uint16_t ports[] = {7010, 7110, 7210, 7310, 7410};
+int32_t  replicaNum = 3;
+int32_t  myIndex = 0;
+
+SRaftId   ids[TSDB_MAX_REPLICA];
+SSyncInfo syncInfo;
+SSyncFSM* pFsm;
+
+SSyncNode* syncNodeInit() {
+  syncInfo.vgId = 1234;
+  syncInfo.rpcClient = gSyncIO->clientRpc;
+  syncInfo.FpSendMsg = syncIOSendMsg;
+  syncInfo.queue = gSyncIO->pMsgQ;
+  syncInfo.FpEqMsg = syncIOEqMsg;
+  syncInfo.pFsm = pFsm;
+  snprintf(syncInfo.path, sizeof(syncInfo.path), "%s", "./");
+
+  SSyncCfg* pCfg = &syncInfo.syncCfg;
+  pCfg->myIndex = myIndex;
+  pCfg->replicaNum = replicaNum;
+
+  for (int i = 0; i < replicaNum; ++i) {
+    pCfg->nodeInfo[i].nodePort = ports[i];
+    snprintf(pCfg->nodeInfo[i].nodeFqdn, sizeof(pCfg->nodeInfo[i].nodeFqdn), "%s", "127.0.0.1");
+    // taosGetFqdn(pCfg->nodeInfo[0].nodeFqdn);
+  }
+
+  SSyncNode* pSyncNode = syncNodeOpen(&syncInfo);
+  assert(pSyncNode != NULL);
+
+  gSyncIO->FpOnSyncPing = pSyncNode->FpOnPing;
+  gSyncIO->FpOnSyncPingReply = pSyncNode->FpOnPingReply;
+  gSyncIO->FpOnSyncClientRequest = pSyncNode->FpOnClientRequest;
+  gSyncIO->FpOnSyncRequestVote = pSyncNode->FpOnRequestVote;
+  gSyncIO->FpOnSyncRequestVoteReply = pSyncNode->FpOnRequestVoteReply;
+  gSyncIO->FpOnSyncAppendEntries = pSyncNode->FpOnAppendEntries;
+  gSyncIO->FpOnSyncAppendEntriesReply = pSyncNode->FpOnAppendEntriesReply;
+  gSyncIO->FpOnSyncTimeout = pSyncNode->FpOnTimeout;
+  gSyncIO->pSyncNode = pSyncNode;
+
+  return pSyncNode;
+}
+
+SSyncNode* syncInitTest() { return syncNodeInit(); }
+
+void initRaftId(SSyncNode* pSyncNode) {
+  for (int i = 0; i < replicaNum; ++i) {
+    ids[i] = pSyncNode->replicasId[i];
+    char* s = syncUtilRaftId2Str(&ids[i]);
+    printf("raftId[%d] : %s\n", i, s);
+    free(s);
+  }
+}
+
+int main(int argc, char** argv) {
+  // taosInitLog((char *)"syncTest.log", 100000, 10);
+  tsAsyncLog = 0;
+  sDebugFlag = 143 + 64;
+
+  myIndex = 0;
+  if (argc >= 2) {
+    myIndex = atoi(argv[1]);
+  }
+
+  int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]);
+  assert(ret == 0);
+
+  ret = syncEnvStart();
+  assert(ret == 0);
+
+  SSyncNode* pSyncNode = syncInitTest();
+  assert(pSyncNode != NULL);
+  syncNodePrint2((char*)"", pSyncNode);
+
+  initRaftId(pSyncNode);
+
+  //---------------------------
+
+  while (1) {
+    syncNodePingSelf(pSyncNode);
+    taosMsleep(1000);
+  }
+
+  return 0;
+}

source/libs/sync/test/syncPingTimerTest.cpp

@@ -47,6 +47,7 @@ SSyncNode* syncNodeInit() {
 
   gSyncIO->FpOnSyncPing = pSyncNode->FpOnPing;
   gSyncIO->FpOnSyncPingReply = pSyncNode->FpOnPingReply;
+  gSyncIO->FpOnSyncClientRequest = pSyncNode->FpOnClientRequest;
   gSyncIO->FpOnSyncRequestVote = pSyncNode->FpOnRequestVote;
   gSyncIO->FpOnSyncRequestVoteReply = pSyncNode->FpOnRequestVoteReply;
   gSyncIO->FpOnSyncAppendEntries = pSyncNode->FpOnAppendEntries;

source/libs/sync/test/syncPingTimerTest2.cpp

+#include <gtest/gtest.h>
+#include <stdio.h>
+#include "syncEnv.h"
+#include "syncIO.h"
+#include "syncInt.h"
+#include "syncRaftStore.h"
+#include "syncUtil.h"
+
+void logTest() {
+  sTrace("--- sync log test: trace");
+  sDebug("--- sync log test: debug");
+  sInfo("--- sync log test: info");
+  sWarn("--- sync log test: warn");
+  sError("--- sync log test: error");
+  sFatal("--- sync log test: fatal");
+}
+
+uint16_t ports[] = {7010, 7110, 7210, 7310, 7410};
+int32_t  replicaNum = 3;
+int32_t  myIndex = 0;
+
+SRaftId   ids[TSDB_MAX_REPLICA];
+SSyncInfo syncInfo;
+SSyncFSM* pFsm;
+
+SSyncNode* syncNodeInit() {
+  syncInfo.vgId = 1234;
+  syncInfo.rpcClient = gSyncIO->clientRpc;
+  syncInfo.FpSendMsg = syncIOSendMsg;
+  syncInfo.queue = gSyncIO->pMsgQ;
+  syncInfo.FpEqMsg = syncIOEqMsg;
+  syncInfo.pFsm = pFsm;
+  snprintf(syncInfo.path, sizeof(syncInfo.path), "%s", "./");
+
+  SSyncCfg* pCfg = &syncInfo.syncCfg;
+  pCfg->myIndex = myIndex;
+  pCfg->replicaNum = replicaNum;
+
+  for (int i = 0; i < replicaNum; ++i) {
+    pCfg->nodeInfo[i].nodePort = ports[i];
+    snprintf(pCfg->nodeInfo[i].nodeFqdn, sizeof(pCfg->nodeInfo[i].nodeFqdn), "%s", "127.0.0.1");
+    // taosGetFqdn(pCfg->nodeInfo[0].nodeFqdn);
+  }
+
+  SSyncNode* pSyncNode = syncNodeOpen(&syncInfo);
+  assert(pSyncNode != NULL);
+
+  gSyncIO->FpOnSyncPing = pSyncNode->FpOnPing;
+  gSyncIO->FpOnSyncPingReply = pSyncNode->FpOnPingReply;
+  gSyncIO->FpOnSyncClientRequest = pSyncNode->FpOnClientRequest;
+  gSyncIO->FpOnSyncRequestVote = pSyncNode->FpOnRequestVote;
+  gSyncIO->FpOnSyncRequestVoteReply = pSyncNode->FpOnRequestVoteReply;
+  gSyncIO->FpOnSyncAppendEntries = pSyncNode->FpOnAppendEntries;
+  gSyncIO->FpOnSyncAppendEntriesReply = pSyncNode->FpOnAppendEntriesReply;
+  gSyncIO->FpOnSyncTimeout = pSyncNode->FpOnTimeout;
+  gSyncIO->pSyncNode = pSyncNode;
+
+  return pSyncNode;
+}
+
+SSyncNode* syncInitTest() { return syncNodeInit(); }
+
+void initRaftId(SSyncNode* pSyncNode) {
+  for (int i = 0; i < replicaNum; ++i) {
+    ids[i] = pSyncNode->replicasId[i];
+    char* s = syncUtilRaftId2Str(&ids[i]);
+    printf("raftId[%d] : %s\n", i, s);
+    free(s);
+  }
+}
+
+int main(int argc, char** argv) {
+  // taosInitLog((char *)"syncTest.log", 100000, 10);
+  tsAsyncLog = 0;
+  sDebugFlag = 143 + 64;
+
+  myIndex = 0;
+  if (argc >= 2) {
+    myIndex = atoi(argv[1]);
+  }
+
+  int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]);
+  assert(ret == 0);
+
+  ret = syncEnvStart();
+  assert(ret == 0);
+
+  SSyncNode* pSyncNode = syncInitTest();
+  assert(pSyncNode != NULL);
+  syncNodePrint2((char*)"", pSyncNode);
+
+  initRaftId(pSyncNode);
+
+  //---------------------------
+
+  sTrace("syncNodeStartPingTimer ...");
+  ret = syncNodeStartPingTimer(pSyncNode);
+  assert(ret == 0);
+
+  while (1) {
+    sTrace("while 1 sleep ...");
+    taosMsleep(1000);
+  }
+
+  return 0;
+}

source/libs/sync/test/syncRaftStoreTest.cpp

@@ -3,6 +3,7 @@
 #include <stdio.h>
 #include "syncIO.h"
 #include "syncInt.h"
+#include "syncUtil.h"
 
 void logTest() {
   sTrace("--- sync log test: trace");
@@ -13,6 +14,21 @@ void logTest() {
   sFatal("--- sync log test: fatal");
 }
 
+uint16_t ports[] = {7010, 7110, 7210, 7310, 7410};
+int32_t  replicaNum = 5;
+int32_t  myIndex = 0;
+SRaftId  ids[TSDB_MAX_REPLICA];
+
+void initRaftId() {
+  for (int i = 0; i < replicaNum; ++i) {
+    ids[i].addr = syncUtilAddr2U64("127.0.0.1", ports[i]);
+    ids[i].vgId = 1234;
+    char* s = syncUtilRaftId2Str(&ids[i]);
+    printf("raftId[%d] : %s\n", i, s);
+    free(s);
+  }
+}
+
 int main() {
   // taosInitLog((char *)"syncTest.log", 100000, 10);
   tsAsyncLog = 0;
@@ -20,23 +36,35 @@ int main() {
 
   logTest();
 
-  SRaftStore *pRaftStore = raftStoreOpen("./raft_store.json");
+  initRaftId();
+
+  SRaftStore* pRaftStore = raftStoreOpen("./test_raft_store.json");
   assert(pRaftStore != NULL);
-  raftStorePrint(pRaftStore);
-
-#if 0
-  pRaftStore->currentTerm = 100;
-  pRaftStore->voteFor.addr = 200;
-  pRaftStore->voteFor.vgId = 300;
-  raftStorePersist(pRaftStore);
-  raftStorePrint(pRaftStore);
-#endif
-
-  ++(pRaftStore->currentTerm);
-  ++(pRaftStore->voteFor.addr);
-  ++(pRaftStore->voteFor.vgId);
-  raftStorePersist(pRaftStore);
-  raftStorePrint(pRaftStore);
+  raftStorePrint2((char*)"==raftStoreOpen==", pRaftStore);
+
+  raftStoreSetTerm(pRaftStore, 100);
+  raftStorePrint2((char*)"==raftStoreSetTerm==", pRaftStore);
+
+  raftStoreVote(pRaftStore, &ids[0]);
+  raftStorePrint2((char*)"==raftStoreVote==", pRaftStore);
+
+  raftStoreClearVote(pRaftStore);
+  raftStorePrint2((char*)"==raftStoreClearVote==", pRaftStore);
+
+  raftStoreVote(pRaftStore, &ids[1]);
+  raftStorePrint2((char*)"==raftStoreVote==", pRaftStore);
+
+  raftStoreNextTerm(pRaftStore);
+  raftStorePrint2((char*)"==raftStoreNextTerm==", pRaftStore);
+
+  raftStoreNextTerm(pRaftStore);
+  raftStorePrint2((char*)"==raftStoreNextTerm==", pRaftStore);
+
+  raftStoreNextTerm(pRaftStore);
+  raftStorePrint2((char*)"==raftStoreNextTerm==", pRaftStore);
+
+  raftStoreNextTerm(pRaftStore);
+  raftStorePrint2((char*)"==raftStoreNextTerm==", pRaftStore);
 
   return 0;
 }

source/libs/transport/src/transCli.c

@@ -30,6 +30,7 @@ typedef struct SCliConn {
   void*        hostThrd;
   SConnBuffer  readBuf;
   void*        data;
+  SArray*      cliMsgs;
   queue        conn;
   uint64_t     expireTime;
   int          hThrdIdx;
@@ -106,6 +107,7 @@ static void cliAsyncCb(uv_async_t* handle);
 static SCliConn* cliCreateConn(SCliThrdObj* thrd);
 static void      cliDestroyConn(SCliConn* pConn, bool clear /*clear tcp handle or not*/);
 static void      cliDestroy(uv_handle_t* handle);
+static void      cliSend(SCliConn* pConn);
 
 // process data read from server, add decompress etc later
 static void cliHandleResp(SCliConn* conn);
@@ -156,8 +158,18 @@ static void         destroyThrdObj(SCliThrdObj* pThrd);
   } while (0)
 #define CONN_NO_PERSIST_BY_APP(conn) ((conn)->persist == false)
 
+#define REQUEST_NO_RESP(msg) ((msg)->noResp == 1)
+
 static void* cliWorkThread(void* arg);
 
+bool cliMaySendCachedMsg(SCliConn* conn) {
+  if (taosArrayGetSize(conn->cliMsgs) > 0) {
+    cliSend(conn);
+    return true;
+  } else {
+    return false;
+  }
+}
 void cliHandleResp(SCliConn* conn) {
   SCliThrdObj* pThrd = conn->hostThrd;
   STrans*      pTransInst = pThrd->pTransInst;
@@ -173,18 +185,18 @@ void cliHandleResp(SCliConn* conn) {
   transMsg.msgType = pHead->msgType;
   transMsg.ahandle = NULL;
 
-  SCliMsg*       pMsg = conn->data;
+  SCliMsg* pMsg = NULL;
+  if (taosArrayGetSize(conn->cliMsgs) > 0) {
+    pMsg = taosArrayGetP(conn->cliMsgs, 0);
+    taosArrayRemove(conn->cliMsgs, 0);
+  }
+
   STransConnCtx* pCtx = pMsg ? pMsg->ctx : NULL;
   if (pMsg == NULL && !CONN_NO_PERSIST_BY_APP(conn)) {
     transMsg.ahandle = pTransInst->mfp ? (*pTransInst->mfp)(pTransInst->parent, transMsg.msgType) : NULL;
   } else {
     transMsg.ahandle = pCtx ? pCtx->ahandle : NULL;
   }
-  // if (rpcMsg.ahandle == NULL) {
-  //  tDebug("%s cli conn %p handle except", CONN_GET_INST_LABEL(conn), conn);
-  //  return;
-  //}
-
   // buf's mem alread translated to transMsg.pCont
   transClearBuffer(&conn->readBuf);
 
@@ -214,12 +226,15 @@ void cliHandleResp(SCliConn* conn) {
     memcpy((char*)pCtx->pRsp, (char*)&transMsg, sizeof(transMsg));
     tsem_post(pCtx->pSem);
   }
+  destroyCmsg(pMsg);
+
+  if (cliMaySendCachedMsg(conn) == true) {
+    return;
+  }
 
   if (CONN_NO_PERSIST_BY_APP(conn)) {
     addConnToPool(pThrd->pool, conn);
   }
-  destroyCmsg(conn->data);
-  conn->data = NULL;
 
   uv_read_start((uv_stream_t*)conn->stream, cliAllocRecvBufferCb, cliRecvCb);
   // start thread's timer of conn pool if not active
@@ -229,7 +244,7 @@ void cliHandleResp(SCliConn* conn) {
 }
 
 void cliHandleExcept(SCliConn* pConn) {
-  if (pConn->data == NULL) {
+  if (taosArrayGetSize(pConn->cliMsgs) == 0) {
     if (pConn->broken == true || CONN_NO_PERSIST_BY_APP(pConn)) {
       transUnrefCliHandle(pConn);
       return;
@@ -238,32 +253,38 @@ void cliHandleExcept(SCliConn* pConn) {
   SCliThrdObj* pThrd = pConn->hostThrd;
   STrans*      pTransInst = pThrd->pTransInst;
 
-  SCliMsg*       pMsg = pConn->data;
-  STransConnCtx* pCtx = pMsg ? pMsg->ctx : NULL;
+  do {
+    SCliMsg* pMsg = NULL;
+    if (taosArrayGetSize(pConn->cliMsgs) > 0) {
+      pMsg = taosArrayGetP(pConn->cliMsgs, 0);
+      taosArrayRemove(pConn->cliMsgs, 0);
+    }
 
-  STransMsg transMsg = {0};
-  transMsg.code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
-  transMsg.msgType = pMsg ? pMsg->msg.msgType + 1 : 0;
-  transMsg.ahandle = NULL;
+    STransConnCtx* pCtx = pMsg ? pMsg->ctx : NULL;
 
-  if (pMsg == NULL && !CONN_NO_PERSIST_BY_APP(pConn)) {
-    transMsg.ahandle = pTransInst->mfp ? (*pTransInst->mfp)(pTransInst->parent, transMsg.msgType) : NULL;
-  } else {
-    transMsg.ahandle = pCtx ? pCtx->ahandle : NULL;
-  }
+    STransMsg transMsg = {0};
+    transMsg.code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
+    transMsg.msgType = pMsg ? pMsg->msg.msgType + 1 : 0;
+    transMsg.ahandle = NULL;
 
-  if (pCtx == NULL || pCtx->pSem == NULL) {
-    tTrace("%s cli conn %p handle resp", pTransInst->label, pConn);
-    (pTransInst->cfp)(pTransInst->parent, &transMsg, NULL);
-  } else {
-    tTrace("%s cli conn(sync) %p handle resp", pTransInst->label, pConn);
-    memcpy((char*)(pCtx->pRsp), (char*)(&transMsg), sizeof(transMsg));
-    tsem_post(pCtx->pSem);
-  }
-  destroyCmsg(pConn->data);
-  pConn->data = NULL;
+    if (pMsg == NULL && !CONN_NO_PERSIST_BY_APP(pConn)) {
+      transMsg.ahandle = pTransInst->mfp ? (*pTransInst->mfp)(pTransInst->parent, transMsg.msgType) : NULL;
+    } else {
+      transMsg.ahandle = pCtx ? pCtx->ahandle : NULL;
+    }
+
+    if (pCtx == NULL || pCtx->pSem == NULL) {
+      tTrace("%s cli conn %p handle resp", pTransInst->label, pConn);
+      (pTransInst->cfp)(pTransInst->parent, &transMsg, NULL);
+    } else {
+      tTrace("%s cli conn(sync) %p handle resp", pTransInst->label, pConn);
+      memcpy((char*)(pCtx->pRsp), (char*)(&transMsg), sizeof(transMsg));
+      tsem_post(pCtx->pSem);
+    }
+    destroyCmsg(pMsg);
+    tTrace("%s cli conn %p start to destroy", CONN_GET_INST_LABEL(pConn), pConn);
+  } while (taosArrayGetSize(pConn->cliMsgs) > 0);
 
-  tTrace("%s cli conn %p start to destroy", CONN_GET_INST_LABEL(pConn), pConn);
   transUnrefCliHandle(pConn);
 }
 
@@ -398,6 +419,7 @@ static SCliConn* cliCreateConn(SCliThrdObj* pThrd) {
 
   conn->writeReq.data = conn;
   conn->connReq.data = conn;
+  conn->cliMsgs = taosArrayInit(2, sizeof(void*));
 
   QUEUE_INIT(&conn->conn);
   conn->hostThrd = pThrd;
@@ -417,32 +439,51 @@ static void cliDestroy(uv_handle_t* handle) {
   SCliConn* conn = handle->data;
   free(conn->ip);
   free(conn->stream);
+  taosArrayDestroy(conn->cliMsgs);
   tTrace("%s cli conn %p destroy successfully", CONN_GET_INST_LABEL(conn), conn);
   free(conn);
 }
-
+static bool cliHandleNoResp(SCliConn* conn) {
+  bool    res = false;
+  SArray* msgs = conn->cliMsgs;
+  if (taosArrayGetSize(msgs) > 0) {
+    SCliMsg* pMsg = taosArrayGetP(msgs, 0);
+    if (REQUEST_NO_RESP(&pMsg->msg)) {
+      taosArrayRemove(msgs, 0);
+      destroyCmsg(pMsg);
+      res = true;
+    }
+    if (res == true) {
+      if (cliMaySendCachedMsg(conn) == false) {
+        SCliThrdObj* thrd = conn->hostThrd;
+        addConnToPool(thrd->pool, conn);
+      }
+    }
+  }
+  return res;
+}
 static void cliSendCb(uv_write_t* req, int status) {
   SCliConn* pConn = req->data;
 
   if (status == 0) {
     tTrace("%s cli conn %p data already was written out", CONN_GET_INST_LABEL(pConn), pConn);
-    SCliMsg* pMsg = pConn->data;
-    if (pMsg == NULL) {
-      return;
-    }
-    destroyUserdata(&pMsg->msg);
   } else {
     tError("%s cli conn %p failed to write: %s", CONN_GET_INST_LABEL(pConn), pConn, uv_err_name(status));
     cliHandleExcept(pConn);
     return;
   }
+  if (cliHandleNoResp(pConn) == true) {
+    tTrace("%s cli conn %p no resp required", CONN_GET_INST_LABEL(pConn), pConn);
+    return;
+  }
   uv_read_start((uv_stream_t*)pConn->stream, cliAllocRecvBufferCb, cliRecvCb);
 }
 
 void cliSend(SCliConn* pConn) {
   CONN_HANDLE_BROKEN(pConn);
 
-  SCliMsg*       pCliMsg = pConn->data;
+  assert(taosArrayGetSize(pConn->cliMsgs) > 0);
+  SCliMsg*       pCliMsg = taosArrayGetP(pConn->cliMsgs, 0);
   STransConnCtx* pCtx = pCliMsg->ctx;
 
   SCliThrdObj* pThrd = pConn->hostThrd;
@@ -472,6 +513,7 @@ void cliSend(SCliConn* pConn) {
     msgLen += sizeof(STransUserMsg);
   }
 
+  pHead->resflag = REQUEST_NO_RESP(pMsg) ? 1 : 0;
   pHead->msgType = pMsg->msgType;
   pHead->msgLen = (int32_t)htonl((uint32_t)msgLen);
 
@@ -480,6 +522,7 @@ void cliSend(SCliConn* pConn) {
          TMSG_INFO(pHead->msgType), inet_ntoa(pConn->addr.sin_addr), ntohs(pConn->addr.sin_port),
          inet_ntoa(pConn->locaddr.sin_addr), ntohs(pConn->locaddr.sin_port));
 
+  pConn->writeReq.data = pConn;
   uv_write(&pConn->writeReq, (uv_stream_t*)pConn->stream, &wb, 1, cliSendCb);
 
   return;
@@ -502,8 +545,8 @@ void cliConnCb(uv_connect_t* req, int status) {
   uv_tcp_getsockname((uv_tcp_t*)pConn->stream, (struct sockaddr*)&pConn->locaddr, &addrlen);
 
   tTrace("%s cli conn %p connect to server successfully", CONN_GET_INST_LABEL(pConn), pConn);
-
   assert(pConn->stream == req->handle);
+
   cliSend(pConn);
 }
 
@@ -521,8 +564,11 @@ static void cliHandleRelease(SCliMsg* pMsg, SCliThrdObj* pThrd) {
   SCliConn* conn = pMsg->msg.handle;
   tDebug("%s cli conn %p release to inst", CONN_GET_INST_LABEL(conn), conn);
 
-  destroyCmsg(pMsg);
-  conn->data = NULL;
+  while (taosArrayGetSize(conn->cliMsgs) > 0) {
+    SCliMsg* pMsg = taosArrayGetP(conn->cliMsgs, 0);
+    destroyCmsg(pMsg);
+    taosArrayRemove(conn->cliMsgs, 0);
+  }
 
   transDestroyBuffer(&conn->readBuf);
   if (conn->persist && T_REF_VAL_GET(conn) >= 2) {
@@ -561,14 +607,19 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
 
   SCliConn* conn = cliGetConn(pMsg, pThrd);
   if (conn != NULL) {
-    conn->data = pMsg;
     conn->hThrdIdx = pCtx->hThrdIdx;
 
+    if (taosArrayGetSize(conn->cliMsgs) > 0) {
+      taosArrayPush(conn->cliMsgs, &pMsg);
+      return;
+    }
+    taosArrayPush(conn->cliMsgs, &pMsg);
     transDestroyBuffer(&conn->readBuf);
     cliSend(conn);
   } else {
     conn = cliCreateConn(pThrd);
-    conn->data = pMsg;
+    taosArrayPush(conn->cliMsgs, &pMsg);
+
     conn->hThrdIdx = pCtx->hThrdIdx;
     conn->ip = strdup(pMsg->ctx->ip);
     conn->port = pMsg->ctx->port;

source/libs/transport/src/transSrv.c

@@ -226,15 +226,22 @@ static void uvHandleReq(SSrvConn* pConn) {
   transMsg.msgType = pHead->msgType;
   transMsg.code = pHead->code;
   transMsg.ahandle = NULL;
-  transMsg.handle = pConn;
+  transMsg.handle = NULL;
 
   transClearBuffer(&pConn->readBuf);
   pConn->inType = pHead->msgType;
-  transRefSrvHandle(pConn);
 
-  tDebug("server conn %p %s received from %s:%d, local info: %s:%d, msg size: %d", pConn, TMSG_INFO(transMsg.msgType),
-         inet_ntoa(pConn->addr.sin_addr), ntohs(pConn->addr.sin_port), inet_ntoa(pConn->locaddr.sin_addr),
-         ntohs(pConn->locaddr.sin_port), transMsg.contLen);
+  if (pHead->resflag == 0) {
+    transRefSrvHandle(pConn);
+    transMsg.handle = pConn;
+    tDebug("server conn %p %s received from %s:%d, local info: %s:%d, msg size: %d", pConn, TMSG_INFO(transMsg.msgType),
+           inet_ntoa(pConn->addr.sin_addr), ntohs(pConn->addr.sin_port), inet_ntoa(pConn->locaddr.sin_addr),
+           ntohs(pConn->locaddr.sin_port), transMsg.contLen);
+  } else {
+    tDebug("server conn %p %s received from %s:%d, local info: %s:%d, msg size: %d, no resp ", pConn,
+           TMSG_INFO(transMsg.msgType), inet_ntoa(pConn->addr.sin_addr), ntohs(pConn->addr.sin_port),
+           inet_ntoa(pConn->locaddr.sin_addr), ntohs(pConn->locaddr.sin_port), transMsg.contLen);
+  }
 
   STrans* pTransInst = (STrans*)p->shandle;
   (*pTransInst->cfp)(pTransInst->parent, &transMsg, NULL);

source/libs/transport/test/transUT.cc

@@ -50,6 +50,7 @@ static void *ConstructArgForSpecificMsgType(void *parent, tmsg_t msgType) {
 }
 // server except
 static bool handleExcept(void *parent, tmsg_t msgType) {
+  //
   return msgType == TDMT_VND_QUERY || msgType == TDMT_VND_FETCH_RSP || msgType == TDMT_VND_RES_READY_RSP;
 }
 typedef void (*CB)(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet);
@@ -360,7 +361,7 @@ TEST_F(TransEnv, cliPersistHandle) {
   tr->SetCliPersistFp(cliPersistHandle);
   SRpcMsg resp = {0};
   for (int i = 0; i < 10; i++) {
-    SRpcMsg req = {.handle = resp.handle};
+    SRpcMsg req = {.handle = resp.handle, .noResp = 0};
     req.msgType = 1;
     req.pCont = rpcMallocCont(10);
     req.contLen = 10;
@@ -447,6 +448,25 @@ TEST_F(TransEnv, srvPersistHandleExcept) {
   // conn broken
   //
 }
+TEST_F(TransEnv, cliPersistHandleExcept) {
+  tr->SetSrvContinueSend(processContinueSend);
+  tr->SetCliPersistFp(cliPersistHandle);
+  SRpcMsg resp = {0};
+  for (int i = 0; i < 5; i++) {
+    SRpcMsg req = {.handle = resp.handle};
+    req.msgType = 1;
+    req.pCont = rpcMallocCont(10);
+    req.contLen = 10;
+    tr->cliSendAndRecv(&req, &resp);
+    if (i > 2) {
+      tr->StopSrv();
+      break;
+    }
+  }
+  taosMsleep(2000);
+  // conn broken
+  //
+}
 
 TEST_F(TransEnv, multiCliPersistHandleExcept) {
   // conn broken
@@ -457,5 +477,15 @@ TEST_F(TransEnv, queryExcept) {
   // query and conn is broken
 }
 TEST_F(TransEnv, noResp) {
+  SRpcMsg resp = {0};
+  for (int i = 0; i < 5; i++) {
+    SRpcMsg req = {.noResp = 1};
+    req.msgType = 1;
+    req.pCont = rpcMallocCont(10);
+    req.contLen = 10;
+    tr->cliSendAndRecv(&req, &resp);
+  }
+  taosMsleep(2000);
+
   // no resp
 }