Merge remote-tracking branch 'origin/3.0' into feature/3.0_liaohj

include/common/taosdef.h

@@ -33,7 +33,7 @@ typedef enum {
   TSDB_SUPER_TABLE = 1,   // super table
   TSDB_CHILD_TABLE = 2,   // table created from super table
   TSDB_NORMAL_TABLE = 3,  // ordinary table
-  TSDB_STREAM_TABLE = 4,  // table created from stream computing
+  TSDB_STREAM_TABLE = 4,  // table created by stream processing
   TSDB_TEMP_TABLE = 5,    // temp table created by nest query
   TSDB_TABLE_MAX = 6
 } ETableType;
@@ -50,7 +50,12 @@ typedef enum {
   TSDB_CHECK_ITEM_MAX
 } ECheckItemType;
 
-typedef enum { TD_ROW_DISCARD_UPDATE = 0, TD_ROW_OVERWRITE_UPDATE = 1, TD_ROW_PARTIAL_UPDATE = 2 } TDUpdateConfig;
+typedef enum {
+  TD_ROW_DISCARD_UPDATE = 0,
+  TD_ROW_OVERWRITE_UPDATE = 1,
+  TD_ROW_PARTIAL_UPDATE = 2,
+} TDUpdateConfig;
+
 typedef enum {
   TSDB_STATIS_OK = 0,    // statis part exist and load successfully
   TSDB_STATIS_NONE = 1,  // statis part not exist
@@ -61,6 +66,12 @@ typedef enum {
   TSDB_SMA_STAT_EXPIRED = 1,  // not ready or expired
 } ETsdbSmaStat;
 
+typedef enum {
+  TSDB_SMA_TYPE_BLOCK = 0,       // Block-wise SMA
+  TSDB_SMA_TYPE_TIME_RANGE = 1,  // Time-range-wise SMA
+  TSDB_SMA_TYPE_ROLLUP = 2,      // Rollup SMA
+} ETsdbSmaType;
+
 extern char *qtypeStr[];
 
 #define TSDB_PORT_HTTP      11

include/common/tcommon.h

@@ -140,6 +140,23 @@ static FORCE_INLINE void* tDecodeDataBlock(const void* buf, SSDataBlock* pBlock)
   return (void*)buf;
 }
 
+static FORCE_INLINE void tDeleteSSDataBlock(SSDataBlock* pBlock) {
+  if (pBlock == NULL) {
+    return;
+  }
+
+  // int32_t numOfOutput = pBlock->info.numOfCols;
+  int32_t sz = taosArrayGetSize(pBlock->pDataBlock);
+  for (int32_t i = 0; i < sz; ++i) {
+    SColumnInfoData* pColInfoData = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, i);
+    tfree(pColInfoData->pData);
+  }
+
+  taosArrayDestroy(pBlock->pDataBlock);
+  tfree(pBlock->pBlockAgg);
+  // tfree(pBlock);
+}
+
 static FORCE_INLINE int32_t tEncodeSMqPollRsp(void** buf, const SMqPollRsp* pRsp) {
   int32_t tlen = 0;
   int32_t sz = 0;
@@ -182,23 +199,6 @@ static FORCE_INLINE void* tDecodeSMqPollRsp(void* buf, SMqPollRsp* pRsp) {
   return buf;
 }
 
-static FORCE_INLINE void tDeleteSSDataBlock(SSDataBlock* pBlock) {
-  if (pBlock == NULL) {
-    return;
-  }
-
-  // int32_t numOfOutput = pBlock->info.numOfCols;
-  int32_t sz = taosArrayGetSize(pBlock->pDataBlock);
-  for (int32_t i = 0; i < sz; ++i) {
-    SColumnInfoData* pColInfoData = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, i);
-    tfree(pColInfoData->pData);
-  }
-
-  taosArrayDestroy(pBlock->pDataBlock);
-  tfree(pBlock->pBlockAgg);
-  // tfree(pBlock);
-}
-
 static FORCE_INLINE void tDeleteSMqConsumeRsp(SMqPollRsp* pRsp) {
   if (pRsp->schemas) {
     if (pRsp->schemas->nCols) {
@@ -208,10 +208,6 @@ static FORCE_INLINE void tDeleteSMqConsumeRsp(SMqPollRsp* pRsp) {
   }
   taosArrayDestroyEx(pRsp->pBlockData, (void (*)(void*))tDeleteSSDataBlock);
   pRsp->pBlockData = NULL;
-  // for (int32_t i = 0; i < taosArrayGetSize(pRsp->pBlockData); i++) {
-  // SSDataBlock* pDataBlock = (SSDataBlock*)taosArrayGet(pRsp->pBlockData, i);
-  // tDeleteSSDataBlock(pDataBlock);
-  //}
 }
 
 //======================================================================================================================

include/common/tmsg.h

@@ -190,7 +190,10 @@ typedef struct SEp {
 
 typedef struct {
   int32_t contLen;
-  int32_t vgId;
+  union {
+    int32_t vgId;
+    int32_t streamTaskId;
+  };
 } SMsgHead;
 
 // Submit message for one table
@@ -1139,6 +1142,17 @@ int32_t tSerializeSCMCreateStreamReq(void* buf, int32_t bufLen, const SCMCreateS
 int32_t tDeserializeSCMCreateStreamReq(void* buf, int32_t bufLen, SCMCreateStreamReq* pReq);
 void    tFreeSCMCreateStreamReq(SCMCreateStreamReq* pReq);
 
+typedef struct {
+  char    name[TSDB_TOPIC_FNAME_LEN];
+  int64_t streamId;
+  char*   sql;
+  char*   executorMsg;
+} SMVCreateStreamReq, SMSCreateStreamReq;
+
+typedef struct {
+  int64_t streamId;
+} SMVCreateStreamRsp, SMSCreateStreamRsp;
+
 typedef struct {
   char   name[TSDB_TOPIC_FNAME_LEN];
   int8_t igExists;

include/dnode/snode/snode.h

@@ -23,6 +23,9 @@
 extern "C" {
 #endif
 
+#define SND_UNIQUE_THREAD_NUM 2
+#define SND_SHARED_THREAD_NUM 2
+
 /* ------------------------ TYPES EXPOSED ------------------------ */
 typedef struct SDnode SDnode;
 typedef struct SSnode SSnode;
@@ -80,6 +83,10 @@ int32_t sndGetLoad(SSnode *pSnode, SSnodeLoad *pLoad);
  */
 int32_t sndProcessMsg(SSnode *pSnode, SRpcMsg *pMsg, SRpcMsg **pRsp);
 
+int32_t sndProcessUMsg(SSnode *pSnode, SRpcMsg *pMsg);
+
+int32_t sndProcessSMsg(SSnode *pSnode, SRpcMsg *pMsg);
+
 /**
  * @brief Drop a snode.
  *

include/libs/transport/trpc.h

@@ -71,6 +71,10 @@ typedef struct SRpcInit {
   // call back to keep conn or not
   bool (*pfp)(void *parent, tmsg_t msgType);
 
+  // to support Send messages multiple times on a link  
+  // 
+  void* (*mfp)(void *parent, tmsg_t msgType);
+
   void *parent;
 } SRpcInit;
 
@@ -89,6 +93,9 @@ void    rpcSendRecv(void *shandle, SEpSet *pEpSet, SRpcMsg *pReq, SRpcMsg *pRsp)
 int     rpcReportProgress(void *pConn, char *pCont, int contLen);
 void    rpcCancelRequest(int64_t rid);
 
+// just release client conn to rpc instance, no close sock
+void rpcReleaseHandle(void *handle);   
+
 void rpcRefHandle(void *handle, int8_t type);
 void rpcUnrefHandle(void *handle, int8_t type);
 

include/os/osLocale.h

@@ -17,12 +17,16 @@
 #define _TD_OS_LOCALE_H_
 
 #include "os.h"
-#include "osString.h"
 
 #ifdef __cplusplus
 extern "C" {
 #endif
 
+// If the error is in a third-party library, place this header file under the third-party library header file.
+#ifndef ALLOW_FORBID_FUNC
+    #define setlocale SETLOCALE_FUNC_TAOS_FORBID
+#endif
+
 char *taosCharsetReplace(char *charsetstr);
 void  taosGetSystemLocale(char *outLocale, char *outCharset);
 void  taosSetSystemLocale(const char *inLocale, const char *inCharSet);

include/util/taoserror.h

@@ -353,6 +353,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_TDB_MESSED_MSG                TAOS_DEF_ERROR_CODE(0, 0x0614)
 #define TSDB_CODE_TDB_IVLD_TAG_VAL              TAOS_DEF_ERROR_CODE(0, 0x0615)
 #define TSDB_CODE_TDB_NO_CACHE_LAST_ROW         TAOS_DEF_ERROR_CODE(0, 0x0616)
+#define TSDB_CODE_TDB_NO_SMA_INDEX_IN_META      TAOS_DEF_ERROR_CODE(0, 0x0617)
 
 // query
 #define TSDB_CODE_QRY_INVALID_QHANDLE           TAOS_DEF_ERROR_CODE(0, 0x0700)

include/util/tuuid.h

+/*
+ * 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 "os.h"
+#include "taoserror.h"
+#include "thash.h"
+
+/**
+ * Generate an non-negative signed 32bit id
+ *+------------+-----+-----------+---------------+
+ *| uid|localIp| PId | timestamp | serial number |
+ *+------------+-----+-----------+---------------+
+ *|  6bit      |6bit |   12bit   |      8bit     |
+ *+------------+-----+-----------+---------------+
+ * @return
+ */
+int32_t tGenIdPI32(void);
+
+/**
+ * Generate an non-negative signed 64bit id
+ *+------------+-----+-----------+---------------+
+ *| uid|localIp| PId | timestamp | serial number |
+ *+------------+-----+-----------+---------------+
+ *| 12bit      |12bit|24bit      |16bit          |
+ *+------------+-----+-----------+---------------+
+ * @return
+ */
+int64_t tGenIdPI64(void);

source/common/src/tmsg.c

@@ -2666,7 +2666,6 @@ int32_t tSerializeSCMCreateStreamReq(void *buf, int32_t bufLen, const SCMCreateS
   if (tEncodeCStr(&encoder, pReq->sql) < 0) return -1;
   if (tEncodeCStr(&encoder, pReq->physicalPlan) < 0) return -1;
   if (tEncodeCStr(&encoder, pReq->logicalPlan) < 0) return -1;
-
   tEndEncode(&encoder);
 
   int32_t tlen = encoder.pos;

source/dnode/mgmt/impl/inc/dndEnv.h

@@ -90,9 +90,11 @@ typedef struct {
   int32_t      refCount;
   int8_t       deployed;
   int8_t       dropped;
+  int8_t       uniqueWorkerInUse;
   SSnode      *pSnode;
   SRWLatch     latch;
-  SDnodeWorker writeWorker;
+  SArray      *uniqueWorkers;  // SArray<SDnodeWorker*>
+  SDnodeWorker sharedWorker;
 } SSnodeMgmt;
 
 typedef struct {
@@ -153,4 +155,4 @@ int32_t dndGetMonitorDiskInfo(SDnode *pDnode, SMonDiskInfo *pInfo);
 }
 #endif
 
-#endif /*_TD_DND_ENV_H_*/
\ No newline at end of file
+#endif /*_TD_DND_ENV_H_*/

source/dnode/mgmt/impl/inc/dndInt.h

@@ -70,4 +70,4 @@ void dndGetStartup(SDnode *pDnode, SStartupReq *pStartup);
 }
 #endif
 
-#endif /*_TD_DND_INT_H_*/
\ No newline at end of file
+#endif /*_TD_DND_INT_H_*/

source/dnode/mgmt/impl/src/dndSnode.c

@@ -19,7 +19,20 @@
 #include "dndTransport.h"
 #include "dndWorker.h"
 
-static void dndProcessSnodeQueue(SDnode *pDnode, SRpcMsg *pMsg);
+typedef struct {
+  int32_t     vgId;
+  int32_t     refCount;
+  int32_t     snVersion;
+  int8_t      dropped;
+  char       *path;
+  SSnode     *pImpl;
+  STaosQueue *pSharedQ;
+  STaosQueue *pUniqueQ;
+} SSnodeObj;
+
+static void dndProcessSnodeSharedQueue(SDnode *pDnode, SRpcMsg *pMsg);
+
+static void dndProcessSnodeUniqueQueue(SDnode *pDnode, STaosQall *qall, int32_t numOfMsgs);
 
 static SSnode *dndAcquireSnode(SDnode *pDnode) {
   SSnodeMgmt *pMgmt = &pDnode->smgmt;
@@ -152,8 +165,21 @@ static int32_t dndWriteSnodeFile(SDnode *pDnode) {
 
 static int32_t dndStartSnodeWorker(SDnode *pDnode) {
   SSnodeMgmt *pMgmt = &pDnode->smgmt;
-  if (dndInitWorker(pDnode, &pMgmt->writeWorker, DND_WORKER_SINGLE, "snode-write", 0, 1, dndProcessSnodeQueue) != 0) {
-    dError("failed to start snode write worker since %s", terrstr());
+  pMgmt->uniqueWorkers = taosArrayInit(0, sizeof(void *));
+  for (int32_t i = 0; i < SND_UNIQUE_THREAD_NUM; i++) {
+    SDnodeWorker *pUniqueWorker = malloc(sizeof(SDnodeWorker));
+    if (pUniqueWorker == NULL) {
+      return -1;
+    }
+    if (dndInitWorker(pDnode, pUniqueWorker, DND_WORKER_MULTI, "snode-unique", 1, 1, dndProcessSnodeSharedQueue) != 0) {
+      dError("failed to start snode unique worker since %s", terrstr());
+      return -1;
+    }
+    taosArrayPush(pMgmt->uniqueWorkers, &pUniqueWorker);
+  }
+  if (dndInitWorker(pDnode, &pMgmt->sharedWorker, DND_WORKER_SINGLE, "snode-shared", SND_SHARED_THREAD_NUM,
+                    SND_SHARED_THREAD_NUM, dndProcessSnodeSharedQueue)) {
+    dError("failed to start snode shared worker since %s", terrstr());
     return -1;
   }
 
@@ -169,9 +195,13 @@ static void dndStopSnodeWorker(SDnode *pDnode) {
 
   while (pMgmt->refCount > 0) {
     taosMsleep(10);
-  } 
+  }
 
-  dndCleanupWorker(&pMgmt->writeWorker);
+  for (int32_t i = 0; i < taosArrayGetSize(pMgmt->uniqueWorkers); i++) {
+    SDnodeWorker *worker = taosArrayGetP(pMgmt->uniqueWorkers, i);
+    dndCleanupWorker(worker);
+  }
+  taosArrayDestroy(pMgmt->uniqueWorkers);
 }
 
 static void dndBuildSnodeOption(SDnode *pDnode, SSnodeOpt *pOption) {
@@ -292,17 +322,36 @@ int32_t dndProcessDropSnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
   }
 }
 
-static void dndProcessSnodeQueue(SDnode *pDnode, SRpcMsg *pMsg) {
+static void dndProcessSnodeUniqueQueue(SDnode *pDnode, STaosQall *qall, int32_t numOfMsgs) {
+  SSnodeMgmt *pMgmt = &pDnode->smgmt;
+  int32_t     code = TSDB_CODE_DND_SNODE_NOT_DEPLOYED;
+
+  SSnode *pSnode = dndAcquireSnode(pDnode);
+  if (pSnode != NULL) {
+    for (int32_t i = 0; i < numOfMsgs; i++) {
+      SRpcMsg *pMsg = NULL;
+      taosGetQitem(qall, (void **)&pMsg);
+
+      sndProcessUMsg(pSnode, pMsg);
+
+      rpcFreeCont(pMsg->pCont);
+      taosFreeQitem(pMsg);
+    }
+  }
+  dndReleaseSnode(pDnode, pSnode);
+}
+
+static void dndProcessSnodeSharedQueue(SDnode *pDnode, SRpcMsg *pMsg) {
   SSnodeMgmt *pMgmt = &pDnode->smgmt;
-  SRpcMsg    *pRsp = NULL;
   int32_t     code = TSDB_CODE_DND_SNODE_NOT_DEPLOYED;
 
   SSnode *pSnode = dndAcquireSnode(pDnode);
   if (pSnode != NULL) {
-    code = sndProcessMsg(pSnode, pMsg, &pRsp);
+    code = sndProcessSMsg(pSnode, pMsg);
   }
   dndReleaseSnode(pDnode, pSnode);
 
+#if 0
   if (pMsg->msgType & 1u) {
     if (pRsp != NULL) {
       pRsp->ahandle = pMsg->ahandle;
@@ -314,11 +363,58 @@ static void dndProcessSnodeQueue(SDnode *pDnode, SRpcMsg *pMsg) {
       rpcSendResponse(&rpcRsp);
     }
   }
+#endif
 
   rpcFreeCont(pMsg->pCont);
   taosFreeQitem(pMsg);
 }
 
+static FORCE_INLINE int32_t dndGetSWIdFromMsg(SRpcMsg *pMsg) {
+  SMsgHead *pHead = pMsg->pCont;
+  pHead->streamTaskId = htonl(pHead->streamTaskId);
+  return pHead->streamTaskId % SND_UNIQUE_THREAD_NUM;
+}
+
+static void dndWriteSnodeMsgToWorkerByMsg(SDnode *pDnode, SRpcMsg *pMsg) {
+  int32_t code = TSDB_CODE_DND_SNODE_NOT_DEPLOYED;
+
+  SSnode *pSnode = dndAcquireSnode(pDnode);
+  if (pSnode != NULL) {
+    int32_t       index = dndGetSWIdFromMsg(pMsg);
+    SDnodeWorker *pWorker = taosArrayGetP(pDnode->smgmt.uniqueWorkers, index);
+    code = dndWriteMsgToWorker(pWorker, pMsg, sizeof(SRpcMsg));
+  }
+
+  dndReleaseSnode(pDnode, pSnode);
+
+  if (code != 0) {
+    if (pMsg->msgType & 1u) {
+      SRpcMsg rsp = {.handle = pMsg->handle, .ahandle = pMsg->ahandle, .code = code};
+      rpcSendResponse(&rsp);
+    }
+    rpcFreeCont(pMsg->pCont);
+  }
+}
+
+static void dndWriteSnodeMsgToMgmtWorker(SDnode *pDnode, SRpcMsg *pMsg) {
+  int32_t code = TSDB_CODE_DND_SNODE_NOT_DEPLOYED;
+
+  SSnode *pSnode = dndAcquireSnode(pDnode);
+  if (pSnode != NULL) {
+    SDnodeWorker *pWorker = taosArrayGet(pDnode->smgmt.uniqueWorkers, 0);
+    code = dndWriteMsgToWorker(pWorker, pMsg, sizeof(SRpcMsg));
+  }
+  dndReleaseSnode(pDnode, pSnode);
+
+  if (code != 0) {
+    if (pMsg->msgType & 1u) {
+      SRpcMsg rsp = {.handle = pMsg->handle, .ahandle = pMsg->ahandle, .code = code};
+      rpcSendResponse(&rsp);
+    }
+    rpcFreeCont(pMsg->pCont);
+  }
+}
+
 static void dndWriteSnodeMsgToWorker(SDnode *pDnode, SDnodeWorker *pWorker, SRpcMsg *pMsg) {
   int32_t code = TSDB_CODE_DND_SNODE_NOT_DEPLOYED;
 
@@ -337,8 +433,16 @@ static void dndWriteSnodeMsgToWorker(SDnode *pDnode, SDnodeWorker *pWorker, SRpc
   }
 }
 
-void dndProcessSnodeWriteMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
-  dndWriteSnodeMsgToWorker(pDnode, &pDnode->smgmt.writeWorker, pMsg);
+void dndProcessSnodeMgmtMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
+  dndWriteSnodeMsgToMgmtWorker(pDnode, pMsg);
+}
+
+void dndProcessSnodeUniqueMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
+  dndWriteSnodeMsgToWorkerByMsg(pDnode, pMsg);
+}
+
+void dndProcessSnodeSharedMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
+  dndWriteSnodeMsgToWorker(pDnode, &pDnode->smgmt.sharedWorker, pMsg);
 }
 
 int32_t dndInitSnode(SDnode *pDnode) {

source/dnode/mgmt/impl/src/dndWorker.c

@@ -109,4 +109,4 @@ int32_t dndWriteMsgToWorker(SDnodeWorker *pWorker, void *pCont, int32_t contLen)
   }
 
   return 0;
-}
\ No newline at end of file
+}

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

@@ -30,6 +30,7 @@
 #include "mndShow.h"
 #include "mndSnode.h"
 #include "mndStb.h"
+#include "mndStream.h"
 #include "mndSubscribe.h"
 #include "mndSync.h"
 #include "mndTelem.h"
@@ -220,6 +221,7 @@ static int32_t mndInitSteps(SMnode *pMnode) {
   if (mndAllocStep(pMnode, "mnode-user", mndInitUser, mndCleanupUser) != 0) return -1;
   if (mndAllocStep(pMnode, "mnode-auth", mndInitAuth, mndCleanupAuth) != 0) return -1;
   if (mndAllocStep(pMnode, "mnode-acct", mndInitAcct, mndCleanupAcct) != 0) return -1;
+  if (mndAllocStep(pMnode, "mnode-stream", mndInitStream, mndCleanupStream) != 0) return -1;
   if (mndAllocStep(pMnode, "mnode-topic", mndInitTopic, mndCleanupTopic) != 0) return -1;
   if (mndAllocStep(pMnode, "mnode-consumer", mndInitConsumer, mndCleanupConsumer) != 0) return -1;
   if (mndAllocStep(pMnode, "mnode-subscribe", mndInitSubscribe, mndCleanupSubscribe) != 0) return -1;

source/dnode/snode/inc/sndInt.h

@@ -30,41 +30,42 @@ extern "C" {
 #endif
 
 enum {
-  STREAM_STATUS__READY = 1,
+  STREAM_STATUS__RUNNING = 1,
   STREAM_STATUS__STOPPED,
   STREAM_STATUS__CREATING,
   STREAM_STATUS__STOPING,
-  STREAM_STATUS__RESUMING,
+  STREAM_STATUS__RESTORING,
   STREAM_STATUS__DELETING,
 };
 
 enum {
-  STREAM_RUNNER__RUNNING = 1,
-  STREAM_RUNNER__STOP,
+  STREAM_TASK_STATUS__RUNNING = 1,
+  STREAM_TASK_STATUS__STOP,
 };
 
+typedef struct {
+  SHashObj* pHash;  // taskId -> streamTask
+} SStreamMeta;
+
 typedef struct SSnode {
-  SSnodeOpt cfg;
+  SStreamMeta* pMeta;
+  SSnodeOpt    cfg;
 } SSnode;
 
 typedef struct {
   int64_t streamId;
+  int32_t taskId;
   int32_t IdxInLevel;
   int32_t level;
-} SStreamInfo;
+} SStreamTaskInfo;
 
 typedef struct {
-  SStreamInfo meta;
-  int8_t      status;
-  void*       executor;
-  STaosQueue* queue;
-  void*       stateStore;
+  SStreamTaskInfo meta;
+  int8_t          status;
+  void*           executor;
+  void*           stateStore;
   // storage handle
-} SStreamRunner;
-
-typedef struct {
-  SHashObj* pHash;
-} SStreamMeta;
+} SStreamTask;
 
 int32_t sndCreateStream();
 int32_t sndDropStream();

source/dnode/snode/src/snode.c

@@ -14,6 +14,7 @@
  */
 
 #include "sndInt.h"
+#include "tuuid.h"
 
 SSnode *sndOpen(const char *path, const SSnodeOpt *pOption) {
   SSnode *pSnode = calloc(1, sizeof(SSnode));
@@ -31,3 +32,25 @@ int32_t sndProcessMsg(SSnode *pSnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
 }
 
 void sndDestroy(const char *path) {}
+
+static int32_t sndDeployTask(SSnode *pSnode, SRpcMsg *pMsg) {
+  SStreamTask *task = malloc(sizeof(SStreamTask));
+  if (task == NULL) {
+    return -1;
+  }
+  task->meta.taskId = tGenIdPI32();
+  taosHashPut(pSnode->pMeta->pHash, &task->meta.taskId, sizeof(int32_t), &task, sizeof(void *));
+  return 0;
+}
+
+int32_t sndProcessUMsg(SSnode *pSnode, SRpcMsg *pMsg) {
+  // stream deployment
+  // stream stop/resume
+  // operator exec
+  return 0;
+}
+
+int32_t sndProcessSMsg(SSnode *pSnode, SRpcMsg *pMsg) {
+  // operator exec
+  return 0;
+}

source/dnode/vnode/inc/tsdb.h

@@ -95,6 +95,7 @@ int    tsdbCommit(STsdb *pTsdb);
  * @return int32_t
  */
 int32_t tsdbInsertTSmaData(STsdb *pTsdb, char *msg);
+int32_t tsdbUpdateSmaWindow(STsdb *pTsdb, int8_t smaType, char *msg);
 
 /**
  * @brief Insert RSma(Time-range-wise Rollup SMA) data.
@@ -105,6 +106,12 @@ int32_t tsdbInsertTSmaData(STsdb *pTsdb, char *msg);
  */
 int32_t tsdbInsertRSmaData(STsdb *pTsdb, char *msg);
 
+// TODO: This is the basic params, and should wrap the params to a queryHandle.
+int32_t tsdbGetTSmaData(STsdb *pTsdb, STSmaDataWrapper *pData, int64_t indexUid, int64_t interval,
+                            int8_t intervalUnit, tb_uid_t tableUid, col_id_t colId, TSKEY querySkey,
+                            int32_t nMaxResult);
+
+
 // STsdbCfg
 int  tsdbOptionsInit(STsdbCfg *);
 void tsdbOptionsClear(STsdbCfg *);

source/dnode/vnode/src/inc/tsdbDBDef.h

+/*
+ * 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/>.
+ */
+
+#ifndef _TD_TSDB_DB_DEF_H_
+#define _TD_TSDB_DB_DEF_H_
+
+#include "db.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef struct SDBFile SDBFile;
+typedef DB_ENV*        TDBEnv;
+
+struct SDBFile {
+  DB*   pDB;
+  char* path;
+};
+
+int32_t tsdbOpenDBF(TDBEnv pEnv, SDBFile* pDBF);
+void    tsdbCloseDBF(SDBFile* pDBF);
+int32_t tsdbOpenBDBEnv(DB_ENV** ppEnv, const char* path);
+void    tsdbCloseBDBEnv(DB_ENV* pEnv);
+int32_t tsdbSaveSmaToDB(SDBFile* pDBF, void* key, uint32_t keySize, void* data, uint32_t dataSize);
+void*   tsdbGetSmaDataByKey(SDBFile* pDBF, void* key, uint32_t keySize, uint32_t* valueSize);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_TSDB_DB_DEF_H_*/

source/dnode/vnode/src/inc/tsdbDef.h

@@ -27,6 +27,7 @@
 #include "ttime.h"
 
 #include "tsdb.h"
+#include "tsdbDBDef.h"
 #include "tsdbCommit.h"
 #include "tsdbFS.h"
 #include "tsdbFile.h"
@@ -37,12 +38,15 @@
 #include "tsdbReadImpl.h"
 #include "tsdbSma.h"
 
+
 #ifdef __cplusplus
 extern "C" {
 #endif
 
 struct STsdb {
   int32_t               vgId;
+  bool                  repoLocked;
+  pthread_mutex_t       mutex;
   char *                path;
   STsdbCfg              config;
   STsdbMemTable *       mem;
@@ -52,12 +56,17 @@ struct STsdb {
   STsdbFS *             fs;
   SMeta *               pMeta;
   STfs *                pTfs;
-  SSmaStat *            pSmaStat;
+  SSmaEnv *             pTSmaEnv;
+  SSmaEnv *             pRSmaEnv;
 };
 
-#define REPO_ID(r) ((r)->vgId)
-#define REPO_CFG(r) (&(r)->config)
-#define REPO_FS(r) (r)->fs
+#define REPO_ID(r)        ((r)->vgId)
+#define REPO_CFG(r)       (&(r)->config)
+#define REPO_FS(r)        (r)->fs
+#define IS_REPO_LOCKED(r) (r)->repoLocked
+
+int tsdbLockRepo(STsdb *pTsdb);
+int tsdbUnlockRepo(STsdb *pTsdb);
 
 static FORCE_INLINE STSchema *tsdbGetTableSchemaImpl(STable *pTable, bool lock, bool copy, int32_t version) {
   return pTable->pSchema;

source/dnode/vnode/src/inc/tsdbFile.h

@@ -329,21 +329,23 @@ static FORCE_INLINE int tsdbCopyDFile(SDFile* pSrc, SDFile* pDest) {
 // =============== SDFileSet
 typedef struct {
   int      fid;
-  int8_t   state;    // -128~127
-  uint8_t  ver;      // 0~255, DFileSet version
+  int8_t   state;  // -128~127
+  uint8_t  ver;    // 0~255, DFileSet version
   uint16_t reserve;
   SDFile   files[TSDB_FILE_MAX];
 } SDFileSet;
 
 typedef struct {
-  int     fid;
-  int8_t  state;
-  uint8_t ver;
+  int      fid;
+  int8_t   state;
+  uint8_t  ver;
+  uint16_t reserve;
 #if 0
   SDFInfo   info;
 #endif
   STfsFile  f;
   TdFilePtr pFile;
+
 } SSFile;  // files split by days with fid
 
 #define TSDB_LATEST_FSET_VER 0

source/dnode/vnode/src/inc/tsdbSma.h

@@ -17,27 +17,29 @@
 #define _TD_TSDB_SMA_H_
 
 typedef struct SSmaStat SSmaStat;
+typedef struct SSmaEnv  SSmaEnv;
 
-// insert/update interface
-int32_t tsdbInsertTSmaDataImpl(STsdb *pTsdb, char *msg);
-int32_t tsdbInsertRSmaDataImpl(STsdb *pTsdb, char *msg);
+struct SSmaEnv {
+  pthread_rwlock_t lock;
+  TDBEnv           dbEnv;
+  char *           path;
+  SSmaStat *       pStat;
+};
 
+#define SMA_ENV_LOCK(env)       ((env)->lock)
+#define SMA_ENV_ENV(env)        ((env)->dbEnv)
+#define SMA_ENV_PATH(env)       ((env)->path)
+#define SMA_ENV_STAT(env)       ((env)->pStat)
+#define SMA_ENV_STAT_ITEMS(env) ((env)->pStat->smaStatItems)
 
-// query interface
-// TODO: This is the basic params, and should wrap the params to a queryHandle.
-int32_t tsdbGetTSmaDataImpl(STsdb *pTsdb, STSmaDataWrapper *pData, STimeWindow *queryWin, int32_t nMaxResult);
-
-// management interface
-int32_t tsdbUpdateExpiredWindow(STsdb *pTsdb, char *msg);
-int32_t tsdbDestroySmaState(SSmaStat *pSmaStat);
+void  tsdbDestroySmaEnv(SSmaEnv *pSmaEnv);
+void *tsdbFreeSmaEnv(SSmaEnv *pSmaEnv);
 #if 0
 int32_t tsdbGetTSmaStatus(STsdb *pTsdb, STSma *param, void *result);
 int32_t tsdbRemoveTSmaData(STsdb *pTsdb, STSma *param, STimeWindow *pWin);
 #endif
 
 // internal func
-
-
 static FORCE_INLINE int32_t tsdbEncodeTSmaKey(tb_uid_t tableUid, col_id_t colId, TSKEY tsKey, void **pData) {
   int32_t len = 0;
   len += taosEncodeFixedI64(pData, tableUid);
@@ -46,4 +48,31 @@ static FORCE_INLINE int32_t tsdbEncodeTSmaKey(tb_uid_t tableUid, col_id_t colId,
   return len;
 }
 
+static FORCE_INLINE int tsdbRLockSma(SSmaEnv *pEnv) {
+  int code = pthread_rwlock_rdlock(&(pEnv->lock));
+  if (code != 0) {
+    terrno = TAOS_SYSTEM_ERROR(code);
+    return -1;
+  }
+  return 0;
+}
+
+static FORCE_INLINE int tsdbWLockSma(SSmaEnv *pEnv) {
+  int code = pthread_rwlock_wrlock(&(pEnv->lock));
+  if (code != 0) {
+    terrno = TAOS_SYSTEM_ERROR(code);
+    return -1;
+  }
+  return 0;
+}
+
+static FORCE_INLINE int tsdbUnLockSma(SSmaEnv *pEnv) {
+  int code = pthread_rwlock_unlock(&(pEnv->lock));
+  if (code != 0) {
+    terrno = TAOS_SYSTEM_ERROR(code);
+    return -1;
+  }
+  return 0;
+}
+
 #endif /* _TD_TSDB_SMA_H_ */
\ No newline at end of file

source/dnode/vnode/src/meta/metaBDBImpl.c

@@ -231,30 +231,31 @@ int metaSaveSmaToDB(SMeta *pMeta, STSma *pSmaCfg) {
   void *pBuf = NULL, *qBuf = NULL;
   DBT   key1 = {0}, value1 = {0};
 
-  {
-    // save sma info
-    int32_t len = tEncodeTSma(NULL, pSmaCfg);
-    pBuf = calloc(len, 1);
-    if (pBuf == NULL) {
-      terrno = TSDB_CODE_OUT_OF_MEMORY;
-      return -1;
-    }
+  // save sma info
+  int32_t len = tEncodeTSma(NULL, pSmaCfg);
+  pBuf = calloc(len, 1);
+  if (pBuf == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return -1;
+  }
 
-    key1.data = (void *)&pSmaCfg->indexUid;
-    key1.size = sizeof(pSmaCfg->indexUid);
+  key1.data = (void *)&pSmaCfg->indexUid;
+  key1.size = sizeof(pSmaCfg->indexUid);
 
-    qBuf = pBuf;
-    tEncodeTSma(&qBuf, pSmaCfg);
+  qBuf = pBuf;
+  tEncodeTSma(&qBuf, pSmaCfg);
 
-    value1.data = pBuf;
-    value1.size = POINTER_DISTANCE(qBuf, pBuf);
-    value1.app_data = pSmaCfg;
-  }
+  value1.data = pBuf;
+  value1.size = POINTER_DISTANCE(qBuf, pBuf);
+  value1.app_data = pSmaCfg;
 
   metaDBWLock(pMeta->pDB);
   pMeta->pDB->pSmaDB->put(pMeta->pDB->pSmaDB, NULL, &key1, &value1, 0);
   metaDBULock(pMeta->pDB);
 
+  // release
+  tfree(pBuf);
+
   return 0;
 }
 

source/dnode/vnode/src/tq/tqRead.c

@@ -83,8 +83,8 @@ bool tqNextDataBlock(STqReadHandle* pHandle) {
 }
 
 int tqRetrieveDataBlockInfo(STqReadHandle* pHandle, SDataBlockInfo* pBlockInfo) {
-  /*int32_t         sversion = pHandle->pBlock->sversion;*/
-  /*SSchemaWrapper* pSchema = metaGetTableSchema(pHandle->pMeta, pHandle->pBlock->uid, sversion, false);*/
+  // currently only rows are used
+
   pBlockInfo->numOfCols = taosArrayGetSize(pHandle->pColIdList);
   pBlockInfo->rows = pHandle->pBlock->numOfRows;
   pBlockInfo->uid = pHandle->pBlock->uid;

source/dnode/vnode/src/tsdb/tsdbBDBImpl.c

@@ -12,3 +12,162 @@
  * 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/>.
  */
+
+#define ALLOW_FORBID_FUNC
+#include "db.h"
+
+#include "taoserror.h"
+#include "tcoding.h"
+#include "thash.h"
+#include "tsdbDBDef.h"
+#include "tsdbLog.h"
+
+#define IMPL_WITH_LOCK 1
+
+static int  tsdbOpenBDBDb(DB **ppDB, DB_ENV *pEnv, const char *pFName, bool isDup);
+static void tsdbCloseBDBDb(DB *pDB);
+
+#define BDB_PERR(info, code) fprintf(stderr, "%s:%d " info " reason: %s\n", __FILE__, __LINE__, db_strerror(code))
+
+int32_t tsdbOpenDBF(TDBEnv pEnv, SDBFile *pDBF) {
+  // TDBEnv is shared by a group of SDBFile
+  if (!pEnv) {
+    terrno = TSDB_CODE_INVALID_PTR;
+    return -1;
+  }
+
+  // Open DBF
+  if (tsdbOpenBDBDb(&(pDBF->pDB), pEnv, pDBF->path, false) < 0) {
+    terrno = TSDB_CODE_TDB_INIT_FAILED;
+    tsdbCloseBDBDb(pDBF->pDB);
+    return -1;
+  }
+
+  return 0;
+}
+
+void tsdbCloseDBF(SDBFile *pDBF) {
+  if (pDBF->pDB) {
+    tsdbCloseBDBDb(pDBF->pDB);
+    pDBF->pDB = NULL;
+  }
+  tfree(pDBF->path);
+}
+
+int32_t tsdbOpenBDBEnv(DB_ENV **ppEnv, const char *path) {
+  int     ret = 0;
+  DB_ENV *pEnv = NULL;
+
+  if (path == NULL) return 0;
+
+  ret = db_env_create(&pEnv, 0);
+  if (ret != 0) {
+    BDB_PERR("Failed to create tsdb env", ret);
+    return -1;
+  }
+
+  ret = pEnv->open(pEnv, path, DB_CREATE | DB_INIT_CDB | DB_INIT_MPOOL, 0);
+  if (ret != 0) {
+    // BDB_PERR("Failed to open tsdb env", ret);
+    tsdbWarn("Failed to open tsdb env for path %s since %d", path ? path : "NULL", ret);
+    return -1;
+  }
+
+  *ppEnv = pEnv;
+
+  return 0;
+}
+
+void tsdbCloseBDBEnv(DB_ENV *pEnv) {
+  if (pEnv) {
+    pEnv->close(pEnv, 0);
+  }
+}
+
+static int tsdbOpenBDBDb(DB **ppDB, DB_ENV *pEnv, const char *pFName, bool isDup) {
+  int ret;
+  DB *pDB;
+
+  ret = db_create(&(pDB), pEnv, 0);
+  if (ret != 0) {
+    BDB_PERR("Failed to create DBP", ret);
+    return -1;
+  }
+
+  if (isDup) {
+    ret = pDB->set_flags(pDB, DB_DUPSORT);
+    if (ret != 0) {
+      BDB_PERR("Failed to set DB flags", ret);
+      return -1;
+    }
+  }
+
+  ret = pDB->open(pDB, NULL, pFName, NULL, DB_BTREE, DB_CREATE, 0);
+  if (ret) {
+    BDB_PERR("Failed to open DBF", ret);
+    return -1;
+  }
+
+  *ppDB = pDB;
+
+  return 0;
+}
+
+static void tsdbCloseBDBDb(DB *pDB) {
+  if (pDB) {
+    pDB->close(pDB, 0);
+  }
+}
+
+int32_t tsdbSaveSmaToDB(SDBFile *pDBF, void *key, uint32_t keySize, void *data, uint32_t dataSize) {
+  int ret;
+  DBT key1 = {0}, value1 = {0};
+
+  key1.data = key;
+  key1.size = keySize;
+
+  value1.data = data;
+  value1.size = dataSize;
+
+  // TODO: lock
+  ret = pDBF->pDB->put(pDBF->pDB, NULL, &key1, &value1, 0);
+  if (ret) {
+    BDB_PERR("Failed to put data to DBF", ret);
+    // TODO: unlock
+    return -1;
+  }
+  // TODO: unlock
+
+  return 0;
+}
+
+void *tsdbGetSmaDataByKey(SDBFile *pDBF, void* key, uint32_t keySize, uint32_t *valueSize) {
+  void *result = NULL;
+  DBT   key1 = {0};
+  DBT   value1 = {0};
+  int   ret;
+
+  // Set key/value
+  key1.data = key;
+  key1.size = keySize;
+
+  // Query
+  // TODO: lock
+  ret = pDBF->pDB->get(pDBF->pDB, NULL, &key1, &value1, 0);
+  // TODO: unlock
+  if (ret != 0) {
+    return NULL;
+  }
+
+  result = calloc(1, value1.size);
+
+  if (result == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return NULL;
+  }
+
+  *valueSize = value1.size;
+  memcpy(result, value1.data, value1.size);
+
+  return result;
+}
\ No newline at end of file

source/dnode/vnode/src/tsdb/tsdbMain.c

@@ -80,6 +80,8 @@ static STsdb *tsdbNew(const char *path, int32_t vgId, const STsdbCfg *pTsdbCfg,
   pTsdb->pmaf = pMAF;
   pTsdb->pMeta = pMeta;
   pTsdb->pTfs = pTfs;
+  pTsdb->pTSmaEnv = NULL;
+  pTsdb->pRSmaEnv = NULL;
 
   pTsdb->fs = tsdbNewFS(pTsdbCfg);
 
@@ -88,8 +90,9 @@ static STsdb *tsdbNew(const char *path, int32_t vgId, const STsdbCfg *pTsdbCfg,
 
 static void tsdbFree(STsdb *pTsdb) {
   if (pTsdb) {
+    tsdbFreeSmaEnv(pTsdb->pRSmaEnv);
+    tsdbFreeSmaEnv(pTsdb->pTSmaEnv);
     tsdbFreeFS(pTsdb->fs);
-    tsdbDestroySmaState(pTsdb->pSmaStat);
     tfree(pTsdb->path);
     free(pTsdb);
   }
@@ -105,6 +108,30 @@ static void tsdbCloseImpl(STsdb *pTsdb) {
   tsdbCloseFS(pTsdb);
   // TODO
 }
+
+int tsdbLockRepo(STsdb *pTsdb) {
+  int code = pthread_mutex_lock(&pTsdb->mutex);
+  if (code != 0) {
+    tsdbError("vgId:%d failed to lock tsdb since %s", REPO_ID(pTsdb), strerror(errno));
+    terrno = TAOS_SYSTEM_ERROR(code);
+    return -1;
+  }
+  pTsdb->repoLocked = true;
+  return 0;
+}
+
+int tsdbUnlockRepo(STsdb *pTsdb) {
+  ASSERT(IS_REPO_LOCKED(pTsdb));
+  pTsdb->repoLocked = false;
+  int code = pthread_mutex_unlock(&pTsdb->mutex);
+  if (code != 0) {
+    tsdbError("vgId:%d failed to unlock tsdb since %s", REPO_ID(pTsdb), strerror(errno));
+    terrno = TAOS_SYSTEM_ERROR(code);
+    return -1;
+  }
+  return 0;
+}
+
 #if 0
 /*
  * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>

source/dnode/vnode/src/tsdb/tsdbWrite.c

@@ -34,6 +34,7 @@ int tsdbInsertData(STsdb *pTsdb, SSubmitReq *pMsg, SSubmitRsp *pRsp) {
   return tsdbMemTableInsert(pTsdb, pTsdb->mem, pMsg, NULL);
 }
 
+#if 0
 /**
  * @brief Insert/Update tSma(Time-range-wise SMA) data from stream computing engine
  * 
@@ -51,6 +52,14 @@ int32_t tsdbInsertTSmaData(STsdb *pTsdb, char *msg) {
   return code;
 }
 
+int32_t tsdbUpdateSmaWindow(STsdb *pTsdb, int8_t smaType, char *msg) {
+  int32_t code = TSDB_CODE_SUCCESS;
+  if ((code = tsdbUpdateExpiredWindow(pTsdb, smaType, msg)) < 0) {
+    tsdbWarn("vgId:%d update expired sma window failed since %s", REPO_ID(pTsdb), tstrerror(terrno));
+  }
+  return code;
+}
+
 /**
  * @brief Insert Time-range-wise Rollup Sma(RSma) data
  *
@@ -65,4 +74,6 @@ int32_t tsdbInsertRSmaData(STsdb *pTsdb, char *msg) {
     tsdbWarn("vgId:%d insert rSma data failed since %s", REPO_ID(pTsdb), tstrerror(terrno));
   }
   return code;
-}
\ No newline at end of file
+}
+
+#endif
\ No newline at end of file

source/dnode/vnode/test/tsdbSmaTest.cpp

@@ -33,7 +33,7 @@ int main(int argc, char **argv) {
   return RUN_ALL_TESTS();
 }
 
-TEST(testCase, tSmaEncodeDecodeTest) {
+TEST(testCase, tSma_Meta_Encode_Decode_Test) {
   // encode
   STSma tSma = {0};
   tSma.version = 0;
@@ -87,8 +87,9 @@ TEST(testCase, tSmaEncodeDecodeTest) {
   tdDestroyTSma(&tSma);
   tdDestroyTSmaWrapper(&dstTSmaWrapper);
 }
+
 #if 1
-TEST(testCase, tSma_DB_Put_Get_Del_Test) {
+TEST(testCase, tSma_metaDB_Put_Get_Del_Test) {
   const char *   smaIndexName1 = "sma_index_test_1";
   const char *   smaIndexName2 = "sma_index_test_2";
   const char *   timezone = "Asia/Shanghai";
@@ -220,13 +221,84 @@ TEST(testCase, tSma_DB_Put_Get_Del_Test) {
 #endif
 
 #if 1
-TEST(testCase, tSmaInsertTest) {
-  const int64_t     indexUid = 2000000002;
+TEST(testCase, tSma_Data_Insert_Query_Test) {
+  // step 1: prepare meta
+  const char *   smaIndexName1 = "sma_index_test_1";
+  const char *   timezone = "Asia/Shanghai";
+  const char *   expr = "select count(a,b, top 20), from table interval 1d, sliding 1h;";
+  const char *   tagsFilter = "where tags.location='Beijing' and tags.district='ChaoYang'";
+  const char *   smaTestDir = "./smaTest";
+  const tb_uid_t tbUid = 1234567890;
+  const int64_t  indexUid1 = 2000000001;
+  const int64_t  interval1 = 1;
+  const int8_t   intervalUnit1 = TD_TIME_UNIT_DAY;
+  const uint32_t nCntTSma = 2;
+  TSKEY          skey1 = 1646987196;
+  const int64_t  testSmaData1 = 100;
+  const int64_t  testSmaData2 = 200;
+  // encode
+  STSma tSma = {0};
+  tSma.version = 0;
+  tSma.intervalUnit = TD_TIME_UNIT_DAY;
+  tSma.interval = 1;
+  tSma.slidingUnit = TD_TIME_UNIT_HOUR;
+  tSma.sliding = 0;
+  tSma.indexUid = indexUid1;
+  tstrncpy(tSma.indexName, smaIndexName1, TSDB_INDEX_NAME_LEN);
+  tstrncpy(tSma.timezone, timezone, TD_TIMEZONE_LEN);
+  tSma.tableUid = tbUid;
+
+  tSma.exprLen = strlen(expr);
+  tSma.expr = (char *)calloc(tSma.exprLen + 1, 1);
+  tstrncpy(tSma.expr, expr, tSma.exprLen + 1);
+
+  tSma.tagsFilterLen = strlen(tagsFilter);
+  tSma.tagsFilter = (char *)calloc(tSma.tagsFilterLen + 1, 1);
+  tstrncpy(tSma.tagsFilter, tagsFilter, tSma.tagsFilterLen + 1);
+
+  SMeta *         pMeta = NULL;
+  STSma *         pSmaCfg = &tSma;
+  const SMetaCfg *pMetaCfg = &defaultMetaOptions;
+
+  taosRemoveDir(smaTestDir);
+
+  pMeta = metaOpen(smaTestDir, pMetaCfg, NULL);
+  assert(pMeta != NULL);
+  // save index 1
+  EXPECT_EQ(metaSaveSmaToDB(pMeta, pSmaCfg), 0);
+
+  // step 2: insert data
   STSmaDataWrapper *pSmaData = NULL;
   STsdb             tsdb = {0};
   STsdbCfg *        pCfg = &tsdb.config;
 
-  pCfg->daysPerFile = 1;
+  tsdb.pMeta = pMeta;
+  tsdb.vgId = 2;
+  tsdb.config.daysPerFile = 10;  // default days is 10
+  tsdb.config.keep1 = 30;
+  tsdb.config.keep2 = 90;
+  tsdb.config.keep = 365;
+  tsdb.config.precision = TSDB_TIME_PRECISION_MILLI;
+  tsdb.config.update = TD_ROW_OVERWRITE_UPDATE;
+  tsdb.config.compression = TWO_STAGE_COMP;
+
+  switch (tsdb.config.precision) {
+    case TSDB_TIME_PRECISION_MILLI:
+      skey1 *= 1e3;
+      break;
+    case TSDB_TIME_PRECISION_MICRO:
+      skey1 *= 1e6;
+      break;
+    case TSDB_TIME_PRECISION_NANO:
+      skey1 *= 1e9;
+      break;
+    default:  // ms
+      skey1 *= 1e3;
+      break;
+  }
+
+  char *msg = (char *)calloc(100, 1);
+  EXPECT_EQ(tsdbUpdateSmaWindow(&tsdb, TSDB_SMA_TYPE_TIME_RANGE, msg), 0);
 
   // init
   int32_t allocCnt = 0;
@@ -235,21 +307,21 @@ TEST(testCase, tSmaInsertTest) {
   void *  buf = NULL;
   EXPECT_EQ(tsdbMakeRoom(&buf, allocStep), 0);
   int32_t  bufSize = taosTSizeof(buf);
-  int32_t  numOfTables = 25;
+  int32_t  numOfTables = 10;
   col_id_t numOfCols = 4096;
   EXPECT_GT(numOfCols, 0);
 
   pSmaData = (STSmaDataWrapper *)buf;
   printf(">> allocate [%d] time to %d and addr is %p\n", ++allocCnt, bufSize, pSmaData);
-  pSmaData->skey = 1646987196;
-  pSmaData->interval = 10;
-  pSmaData->intervalUnit = TD_TIME_UNIT_MINUTE;
-  pSmaData->indexUid = indexUid;
+  pSmaData->skey = skey1;
+  pSmaData->interval = interval1;
+  pSmaData->intervalUnit = intervalUnit1;
+  pSmaData->indexUid = indexUid1;
 
   int32_t len = sizeof(STSmaDataWrapper);
   for (int32_t t = 0; t < numOfTables; ++t) {
     STSmaTbData *pTbData = (STSmaTbData *)POINTER_SHIFT(pSmaData, len);
-    pTbData->tableUid = t;
+    pTbData->tableUid = tbUid + t;
 
     int32_t tableDataLen = sizeof(STSmaTbData);
     for (col_id_t c = 0; c < numOfCols; ++c) {
@@ -262,8 +334,17 @@ TEST(testCase, tSmaInsertTest) {
       }
       STSmaColData *pColData = (STSmaColData *)POINTER_SHIFT(pSmaData, len + tableDataLen);
       pColData->colId = c + PRIMARYKEY_TIMESTAMP_COL_ID;
-      pColData->blockSize = ((c & 1) == 0) ? 8 : 16;
+
       // TODO: fill col data
+      if ((c & 1) == 0) {
+        pColData->blockSize = 8;
+        memcpy(pColData->data, &testSmaData1, 8);
+      } else {
+        pColData->blockSize = 16;
+        memcpy(pColData->data, &testSmaData1, 8);
+        memcpy(POINTER_SHIFT(pColData->data, 8), &testSmaData2, 8);
+      }
+
       tableDataLen += (sizeof(STSmaColData) + pColData->blockSize);
     }
     pTbData->dataLen = (tableDataLen - sizeof(STSmaTbData));
@@ -277,8 +358,24 @@ TEST(testCase, tSmaInsertTest) {
   // execute
   EXPECT_EQ(tsdbInsertTSmaData(&tsdb, (char *)pSmaData), TSDB_CODE_SUCCESS);
 
-  // release
+  // step 3: query
+  uint32_t checkDataCnt = 0;
+  for (int32_t t = 0; t < numOfTables; ++t) {
+    for (col_id_t c = 0; c < numOfCols; ++c) {
+      EXPECT_EQ(tsdbGetTSmaData(&tsdb, NULL, indexUid1, interval1, intervalUnit1, tbUid + t,
+                                c + PRIMARYKEY_TIMESTAMP_COL_ID, skey1, 1),
+                TSDB_CODE_SUCCESS);
+      ++checkDataCnt;
+    }
+  }
+  
+  printf("%s:%d The sma data check count for insert and query is %" PRIu32 "\n", __FILE__, __LINE__, checkDataCnt);
+
+  // release data
   taosTZfree(buf);
+  // release meta
+  tdDestroyTSma(&tSma);
+  metaClose(pMeta);
 }
 #endif
 

source/libs/sync/inc/syncElection.h

@@ -39,7 +39,6 @@ extern "C" {
 //    /\ UNCHANGED <<serverVars, candidateVars, leaderVars, logVars>>
 //
 int32_t syncNodeRequestVotePeers(SSyncNode* pSyncNode);
-
 int32_t syncNodeElect(SSyncNode* pSyncNode);
 int32_t syncNodeRequestVote(SSyncNode* pSyncNode, const SRaftId* destRaftId, const SyncRequestVote* pMsg);
 

source/libs/sync/inc/syncEnv.h

@@ -29,6 +29,7 @@ extern "C" {
 #include "ttimer.h"
 
 #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
@@ -38,17 +39,28 @@ extern "C" {
 #define EMPTY_RAFT_ID ((SRaftId){.addr = 0, .vgId = 0})
 
 typedef struct SSyncEnv {
-  tmr_h pEnvTickTimer;
+  // tick timer
+  tmr_h             pEnvTickTimer;
+  int32_t           envTickTimerMS;
+  uint64_t          envTickTimerLogicClock;  // if use queue, should pass logic clock into queue item
+  uint64_t          envTickTimerLogicClockUser;
+  TAOS_TMR_CALLBACK FpEnvTickTimer;  // Timer Fp
+  uint64_t          envTickTimerCounter;
+
+  // timer manager
   tmr_h pTimerManager;
-  char  name[128];
+
+  // other resources shared by SyncNodes
+  // ...
+
 } SSyncEnv;
 
 extern SSyncEnv* gSyncEnv;
 
 int32_t syncEnvStart();
 int32_t syncEnvStop();
-tmr_h   syncEnvStartTimer(TAOS_TMR_CALLBACK fp, int mseconds, void* param);
-void    syncEnvStopTimer(tmr_h* pTimer);
+int32_t syncEnvStartTimer();
+int32_t syncEnvStopTimer();
 
 #ifdef __cplusplus
 }

source/libs/sync/inc/syncIO.h

@@ -29,22 +29,28 @@ extern "C" {
 #include "tqueue.h"
 #include "trpc.h"
 
+#define TICK_Q_TIMER_MS 1000
+#define TICK_Ping_TIMER_MS 1000
+
 typedef struct SSyncIO {
   STaosQueue *pMsgQ;
-  STaosQset * pQset;
+  STaosQset  *pQset;
   pthread_t   consumerTid;
 
-  void * serverRpc;
-  void * clientRpc;
+  void  *serverRpc;
+  void  *clientRpc;
   SEpSet myAddr;
 
-  void *ioTimerTickQ;
-  void *ioTimerTickPing;
-  void *ioTimerManager;
+  tmr_h   qTimer;
+  int32_t qTimerMS;
+  tmr_h   pingTimer;
+  int32_t pingTimerMS;
+  tmr_h   timerMgr;
 
   void *pSyncNode;
   int32_t (*FpOnSyncPing)(SSyncNode *pSyncNode, SyncPing *pMsg);
   int32_t (*FpOnSyncPingReply)(SSyncNode *pSyncNode, SyncPingReply *pMsg);
+  int32_t (*FpOnSyncClientRequest)(SSyncNode *pSyncNode, SyncClientRequest *pMsg);
   int32_t (*FpOnSyncRequestVote)(SSyncNode *pSyncNode, SyncRequestVote *pMsg);
   int32_t (*FpOnSyncRequestVoteReply)(SSyncNode *pSyncNode, SyncRequestVoteReply *pMsg);
   int32_t (*FpOnSyncAppendEntries)(SSyncNode *pSyncNode, SyncAppendEntries *pMsg);
@@ -59,11 +65,14 @@ extern SSyncIO *gSyncIO;
 
 int32_t syncIOStart(char *host, uint16_t port);
 int32_t syncIOStop();
-int32_t syncIOTickQ();
-int32_t syncIOTickPing();
 int32_t syncIOSendMsg(void *clientRpc, const SEpSet *pEpSet, SRpcMsg *pMsg);
 int32_t syncIOEqMsg(void *queue, SRpcMsg *pMsg);
 
+int32_t syncIOQTimerStart();
+int32_t syncIOQTimerStop();
+int32_t syncIOPingTimerStart();
+int32_t syncIOPingTimerStop();
+
 #ifdef __cplusplus
 }
 #endif

source/libs/sync/inc/syncInt.h

@@ -67,12 +67,12 @@ extern "C" {
     }                                                                   \
   }
 
-struct SRaft;
-typedef struct SRaft SRaft;
-
 struct SyncTimeout;
 typedef struct SyncTimeout SyncTimeout;
 
+struct SyncClientRequest;
+typedef struct SyncClientRequest SyncClientRequest;
+
 struct SyncPing;
 typedef struct SyncPing SyncPing;
 
@@ -117,8 +117,10 @@ typedef struct SSyncNode {
   SSyncCfg    syncCfg;
   char        path[TSDB_FILENAME_LEN];
   char        raftStorePath[TSDB_FILENAME_LEN * 2];
-  SWal*       pWal;
-  void*       rpcClient;
+
+  // sync io
+  SWal* pWal;
+  void* rpcClient;
   int32_t (*FpSendMsg)(void* rpcClient, const SEpSet* pEpSet, SRpcMsg* pMsg);
   void* queue;
   int32_t (*FpEqMsg)(void* queue, SRpcMsg* pMsg);
@@ -164,7 +166,7 @@ typedef struct SSyncNode {
   int32_t           pingTimerMS;
   uint64_t          pingTimerLogicClock;
   uint64_t          pingTimerLogicClockUser;
-  TAOS_TMR_CALLBACK FpPingTimer;  // Timer Fp
+  TAOS_TMR_CALLBACK FpPingTimerCB;  // Timer Fp
   uint64_t          pingTimerCounter;
 
   // elect timer
@@ -172,7 +174,7 @@ typedef struct SSyncNode {
   int32_t           electTimerMS;
   uint64_t          electTimerLogicClock;
   uint64_t          electTimerLogicClockUser;
-  TAOS_TMR_CALLBACK FpElectTimer;  // Timer Fp
+  TAOS_TMR_CALLBACK FpElectTimerCB;  // Timer Fp
   uint64_t          electTimerCounter;
 
   // heartbeat timer
@@ -180,12 +182,13 @@ typedef struct SSyncNode {
   int32_t           heartbeatTimerMS;
   uint64_t          heartbeatTimerLogicClock;
   uint64_t          heartbeatTimerLogicClockUser;
-  TAOS_TMR_CALLBACK FpHeartbeatTimer;  // Timer Fp
+  TAOS_TMR_CALLBACK FpHeartbeatTimerCB;  // Timer Fp
   uint64_t          heartbeatTimerCounter;
 
   // callback
   int32_t (*FpOnPing)(SSyncNode* ths, SyncPing* pMsg);
   int32_t (*FpOnPingReply)(SSyncNode* ths, SyncPingReply* pMsg);
+  int32_t (*FpOnClientRequest)(SSyncNode* ths, SyncClientRequest* pMsg);
   int32_t (*FpOnRequestVote)(SSyncNode* ths, SyncRequestVote* pMsg);
   int32_t (*FpOnRequestVoteReply)(SSyncNode* ths, SyncRequestVoteReply* pMsg);
   int32_t (*FpOnAppendEntries)(SSyncNode* ths, SyncAppendEntries* pMsg);
@@ -194,26 +197,47 @@ typedef struct SSyncNode {
 
 } SSyncNode;
 
+// open/close --------------
 SSyncNode* syncNodeOpen(const SSyncInfo* pSyncInfo);
 void       syncNodeClose(SSyncNode* pSyncNode);
 
-int32_t syncNodeSendMsgById(const SRaftId* destRaftId, SSyncNode* pSyncNode, SRpcMsg* pMsg);
-int32_t syncNodeSendMsgByInfo(const SNodeInfo* nodeInfo, SSyncNode* pSyncNode, SRpcMsg* pMsg);
+// ping --------------
 int32_t syncNodePing(SSyncNode* pSyncNode, const SRaftId* destRaftId, SyncPing* pMsg);
-int32_t syncNodePingAll(SSyncNode* pSyncNode);
-int32_t syncNodePingPeers(SSyncNode* pSyncNode);
 int32_t syncNodePingSelf(SSyncNode* pSyncNode);
+int32_t syncNodePingPeers(SSyncNode* pSyncNode);
+int32_t syncNodePingAll(SSyncNode* pSyncNode);
 
+// timer control --------------
 int32_t syncNodeStartPingTimer(SSyncNode* pSyncNode);
 int32_t syncNodeStopPingTimer(SSyncNode* pSyncNode);
 int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode, int32_t ms);
 int32_t syncNodeStopElectTimer(SSyncNode* pSyncNode);
 int32_t syncNodeRestartElectTimer(SSyncNode* pSyncNode, int32_t ms);
+int32_t syncNodeResetElectTimer(SSyncNode* pSyncNode);
 int32_t syncNodeStartHeartbeatTimer(SSyncNode* pSyncNode);
 int32_t syncNodeStopHeartbeatTimer(SSyncNode* pSyncNode);
+
+// utils --------------
+int32_t syncNodeSendMsgById(const SRaftId* destRaftId, SSyncNode* pSyncNode, SRpcMsg* pMsg);
+int32_t syncNodeSendMsgByInfo(const SNodeInfo* nodeInfo, SSyncNode* pSyncNode, SRpcMsg* pMsg);
 cJSON*  syncNode2Json(const SSyncNode* pSyncNode);
 char*   syncNode2Str(const SSyncNode* pSyncNode);
 
+// raft state change --------------
+void syncNodeUpdateTerm(SSyncNode* pSyncNode, SyncTerm term);
+void syncNodeBecomeFollower(SSyncNode* pSyncNode);
+void syncNodeBecomeLeader(SSyncNode* pSyncNode);
+
+void syncNodeCandidate2Leader(SSyncNode* pSyncNode);
+void syncNodeFollower2Candidate(SSyncNode* pSyncNode);
+void syncNodeLeader2Follower(SSyncNode* pSyncNode);
+void syncNodeCandidate2Follower(SSyncNode* pSyncNode);
+
+// raft vote --------------
+void syncNodeVoteForTerm(SSyncNode* pSyncNode, SyncTerm term, SRaftId* pRaftId);
+void syncNodeVoteForSelf(SSyncNode* pSyncNode);
+void syncNodeMaybeAdvanceCommitIndex(SSyncNode* pSyncNode);
+
 // for debug --------------
 void syncNodePrint(SSyncNode* pObj);
 void syncNodePrint2(char* s, SSyncNode* pObj);

source/libs/sync/inc/syncMessage.h

@@ -39,6 +39,7 @@ typedef enum ESyncMessageType {
   SYNC_REQUEST_VOTE_REPLY = 111,
   SYNC_APPEND_ENTRIES = 113,
   SYNC_APPEND_ENTRIES_REPLY = 115,
+  SYNC_RESPONSE = 119,
 
 } ESyncMessageType;
 
@@ -195,7 +196,7 @@ typedef struct SyncRequestVote {
   SRaftId  srcId;
   SRaftId  destId;
   // private data
-  SyncTerm  currentTerm;
+  SyncTerm  term;
   SyncIndex lastLogIndex;
   SyncTerm  lastLogTerm;
 } SyncRequestVote;
@@ -254,6 +255,7 @@ typedef struct SyncAppendEntries {
   SRaftId  srcId;
   SRaftId  destId;
   // private data
+  SyncTerm  term;
   SyncIndex prevLogIndex;
   SyncTerm  prevLogTerm;
   SyncIndex commitIndex;
@@ -286,6 +288,7 @@ typedef struct SyncAppendEntriesReply {
   SRaftId  srcId;
   SRaftId  destId;
   // private data
+  SyncTerm  term;
   bool      success;
   SyncIndex matchIndex;
 } SyncAppendEntriesReply;

source/libs/sync/inc/syncRaftLog.h

@@ -27,6 +27,9 @@ extern "C" {
 #include "syncRaftEntry.h"
 #include "taosdef.h"
 
+#define SYNC_INDEX_BEGIN 0
+#define SYNC_INDEX_INVALID -1
+
 typedef struct SSyncLogStoreData {
   SSyncNode* pSyncNode;
   SWal*      pWal;

source/libs/sync/inc/syncRaftStore.h

@@ -43,6 +43,12 @@ 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);
+
 // for debug -------------------
 void raftStorePrint(SRaftStore *pObj);
 void raftStorePrint2(char *s, SRaftStore *pObj);

source/libs/sync/inc/syncReplication.h

@@ -52,7 +52,6 @@ extern "C" {
 //    /\ UNCHANGED <<serverVars, candidateVars, leaderVars, logVars>>
 //
 int32_t syncNodeAppendEntriesPeers(SSyncNode* pSyncNode);
-
 int32_t syncNodeReplicate(SSyncNode* pSyncNode);
 int32_t syncNodeAppendEntries(SSyncNode* pSyncNode, const SRaftId* destRaftId, const SyncAppendEntries* pMsg);
 

source/libs/sync/inc/syncUtil.h

@@ -34,6 +34,7 @@ void     syncUtilnodeInfo2EpSet(const SNodeInfo* pNodeInfo, SEpSet* pEpSet);
 void     syncUtilraftId2EpSet(const SRaftId* raftId, SEpSet* pEpSet);
 void     syncUtilnodeInfo2raftId(const SNodeInfo* pNodeInfo, SyncGroupId vgId, SRaftId* raftId);
 bool     syncUtilSameId(const SRaftId* pId1, const SRaftId* pId2);
+bool     syncUtilEmptyId(const SRaftId* pId);
 
 // ---- SSyncBuffer ----
 void syncUtilbufBuild(SSyncBuffer* syncBuf, size_t len);
@@ -52,6 +53,8 @@ const char* syncUtilState2String(ESyncState state);
 bool        syncUtilCanPrint(char c);
 char*       syncUtilprintBin(char* ptr, uint32_t len);
 char*       syncUtilprintBin2(char* ptr, uint32_t len);
+SyncIndex   syncUtilMinIndex(SyncIndex a, SyncIndex b);
+SyncIndex   syncUtilMaxIndex(SyncIndex a, SyncIndex b);
 
 #ifdef __cplusplus
 }

source/libs/sync/src/syncAppendEntries.c

@@ -14,6 +14,11 @@
  */
 
 #include "syncAppendEntries.h"
+#include "syncInt.h"
+#include "syncRaftLog.h"
+#include "syncRaftStore.h"
+#include "syncUtil.h"
+#include "syncVoteMgr.h"
 
 // TLA+ Spec
 // HandleAppendEntriesRequest(i, j, m) ==
@@ -80,4 +85,121 @@
 //                       /\ UNCHANGED <<serverVars, commitIndex, messages>>
 //       /\ UNCHANGED <<candidateVars, leaderVars>>
 //
-int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {}
+int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
+  int32_t ret = 0;
+  syncAppendEntriesLog2("==syncNodeOnAppendEntriesCb==", pMsg);
+
+  if (pMsg->term > ths->pRaftStore->currentTerm) {
+    syncNodeUpdateTerm(ths, pMsg->term);
+  }
+  assert(pMsg->term <= ths->pRaftStore->currentTerm);
+
+  if (pMsg->term == ths->pRaftStore->currentTerm) {
+    ths->leaderCache = pMsg->srcId;
+    syncNodeResetElectTimer(ths);
+  }
+  assert(pMsg->dataLen >= 0);
+
+  SyncTerm localPreLogTerm = 0;
+  if (pMsg->prevLogTerm >= SYNC_INDEX_BEGIN && pMsg->prevLogTerm <= ths->pLogStore->getLastIndex(ths->pLogStore)) {
+    SSyncRaftEntry* pEntry = logStoreGetEntry(ths->pLogStore, pMsg->prevLogTerm);
+    assert(pEntry != NULL);
+    localPreLogTerm = pEntry->term;
+    syncEntryDestory(pEntry);
+  }
+
+  bool logOK =
+      (pMsg->prevLogIndex == SYNC_INDEX_INVALID) ||
+      ((pMsg->prevLogIndex >= SYNC_INDEX_BEGIN) &&
+       (pMsg->prevLogIndex <= ths->pLogStore->getLastIndex(ths->pLogStore)) && (pMsg->prevLogIndex == localPreLogTerm));
+
+  // reject
+  if ((pMsg->term < ths->pRaftStore->currentTerm) ||
+      ((pMsg->term == ths->pRaftStore->currentTerm) && (ths->state == TAOS_SYNC_STATE_FOLLOWER) && !logOK)) {
+    SyncAppendEntriesReply* pReply = syncAppendEntriesReplyBuild();
+    pReply->srcId = ths->myRaftId;
+    pReply->destId = pMsg->srcId;
+    pReply->term = ths->pRaftStore->currentTerm;
+    pReply->success = false;
+    pReply->matchIndex = SYNC_INDEX_INVALID;
+
+    SRpcMsg rpcMsg;
+    syncAppendEntriesReply2RpcMsg(pReply, &rpcMsg);
+    syncNodeSendMsgById(&pReply->destId, ths, &rpcMsg);
+    syncAppendEntriesReplyDestroy(pReply);
+
+    return ret;
+  }
+
+  // return to follower state
+  if (pMsg->term == ths->pRaftStore->currentTerm && ths->state == TAOS_SYNC_STATE_CANDIDATE) {
+    syncNodeBecomeFollower(ths);
+  }
+
+  // accept request
+  if (pMsg->term == ths->pRaftStore->currentTerm && ths->state == TAOS_SYNC_STATE_FOLLOWER && logOK) {
+    bool matchSuccess = false;
+    if (pMsg->prevLogIndex == SYNC_INDEX_INVALID &&
+        ths->pLogStore->getLastIndex(ths->pLogStore) == SYNC_INDEX_INVALID) {
+      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) {
+        matchSuccess = true;
+      }
+      syncEntryDestory(pEntry);
+    }
+
+    if (matchSuccess) {
+      // delete conflict entries
+      if (ths->pLogStore->getLastIndex(ths->pLogStore) > pMsg->prevLogIndex) {
+        SyncIndex fromIndex = pMsg->prevLogIndex + 1;
+        ths->pLogStore->truncate(ths->pLogStore, fromIndex);
+      }
+
+      // append one entry
+      if (pMsg->dataLen > 0) {
+        SSyncRaftEntry* pEntry = syncEntryDeserialize(pMsg->data, pMsg->dataLen);
+        ths->pLogStore->appendEntry(ths->pLogStore, pEntry);
+        syncEntryDestory(pEntry);
+      }
+
+      SyncAppendEntriesReply* pReply = syncAppendEntriesReplyBuild();
+      pReply->srcId = ths->myRaftId;
+      pReply->destId = pMsg->srcId;
+      pReply->term = ths->pRaftStore->currentTerm;
+      pReply->success = true;
+      pReply->matchIndex = pMsg->prevLogIndex + 1;
+
+      SRpcMsg rpcMsg;
+      syncAppendEntriesReply2RpcMsg(pReply, &rpcMsg);
+      syncNodeSendMsgById(&pReply->destId, ths, &rpcMsg);
+
+      syncAppendEntriesReplyDestroy(pReply);
+    } else {
+      SyncAppendEntriesReply* pReply = syncAppendEntriesReplyBuild();
+      pReply->srcId = ths->myRaftId;
+      pReply->destId = pMsg->srcId;
+      pReply->term = ths->pRaftStore->currentTerm;
+      pReply->success = false;
+      pReply->matchIndex = SYNC_INDEX_INVALID;
+
+      SRpcMsg rpcMsg;
+      syncAppendEntriesReply2RpcMsg(pReply, &rpcMsg);
+      syncNodeSendMsgById(&pReply->destId, ths, &rpcMsg);
+      syncAppendEntriesReplyDestroy(pReply);
+    }
+
+    if (pMsg->commitIndex > ths->commitIndex) {
+      if (pMsg->commitIndex <= ths->pLogStore->getLastIndex(ths->pLogStore)) {
+        // commit
+        ths->commitIndex = pMsg->commitIndex;
+        ths->pLogStore->updateCommitIndex(ths->pLogStore, ths->commitIndex);
+      }
+    }
+  }
+
+  return ret;
+}

source/libs/sync/src/syncAppendEntriesReply.c

@@ -14,6 +14,12 @@
  */
 
 #include "syncAppendEntriesReply.h"
+#include "syncIndexMgr.h"
+#include "syncInt.h"
+#include "syncRaftLog.h"
+#include "syncRaftStore.h"
+#include "syncUtil.h"
+#include "syncVoteMgr.h"
 
 // TLA+ Spec
 // HandleAppendEntriesResponse(i, j, m) ==
@@ -28,4 +34,41 @@
 //    /\ Discard(m)
 //    /\ UNCHANGED <<serverVars, candidateVars, logVars, elections>>
 //
-int32_t syncNodeOnAppendEntriesReplyCb(SSyncNode* ths, SyncAppendEntriesReply* pMsg) {}
+int32_t syncNodeOnAppendEntriesReplyCb(SSyncNode* ths, SyncAppendEntriesReply* pMsg) {
+  int32_t ret = 0;
+  syncAppendEntriesReplyLog2("==syncNodeOnAppendEntriesReplyCb==", pMsg);
+
+  if (pMsg->term < ths->pRaftStore->currentTerm) {
+    sTrace("DropStaleResponse, receive term:%lu, current term:%lu", pMsg->term, ths->pRaftStore->currentTerm);
+    return ret;
+  }
+
+  // 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);
+  //  }
+
+  assert(pMsg->term == ths->pRaftStore->currentTerm);
+
+  if (pMsg->success) {
+    // nextIndex = reply.matchIndex + 1
+    syncIndexMgrSetIndex(ths->pNextIndex, &(pMsg->srcId), pMsg->matchIndex + 1);
+
+    // matchIndex = reply.matchIndex
+    syncIndexMgrSetIndex(ths->pMatchIndex, &(pMsg->srcId), pMsg->matchIndex);
+
+    // maybe commit
+    syncNodeMaybeAdvanceCommitIndex(ths);
+
+  } else {
+    SyncIndex nextIndex = syncIndexMgrGetIndex(ths->pNextIndex, &(pMsg->srcId));
+    if (nextIndex > SYNC_INDEX_BEGIN) {
+      --nextIndex;
+    } else {
+      nextIndex = SYNC_INDEX_BEGIN;
+    }
+    syncIndexMgrSetIndex(ths->pNextIndex, &(pMsg->srcId), nextIndex);
+  }
+
+  return ret;
+}

source/libs/sync/src/syncElection.c

@@ -16,6 +16,7 @@
 #include "syncElection.h"
 #include "syncMessage.h"
 #include "syncRaftStore.h"
+#include "syncVoteMgr.h"
 
 // TLA+ Spec
 // RequestVote(i, j) ==
@@ -37,7 +38,7 @@ int32_t syncNodeRequestVotePeers(SSyncNode* pSyncNode) {
     SyncRequestVote* pMsg = syncRequestVoteBuild();
     pMsg->srcId = pSyncNode->myRaftId;
     pMsg->destId = pSyncNode->peersId[i];
-    pMsg->currentTerm = pSyncNode->pRaftStore->currentTerm;
+    pMsg->term = pSyncNode->pRaftStore->currentTerm;
     pMsg->lastLogIndex = pSyncNode->pLogStore->getLastIndex(pSyncNode->pLogStore);
     pMsg->lastLogTerm = pSyncNode->pLogStore->getLastTerm(pSyncNode->pLogStore);
 
@@ -49,10 +50,22 @@ int32_t syncNodeRequestVotePeers(SSyncNode* pSyncNode) {
 }
 
 int32_t syncNodeElect(SSyncNode* pSyncNode) {
+  if (pSyncNode->state == TAOS_SYNC_STATE_FOLLOWER) {
+    syncNodeFollower2Candidate(pSyncNode);
+  }
   assert(pSyncNode->state == TAOS_SYNC_STATE_CANDIDATE);
 
   // start election
+  raftStoreNextTerm(pSyncNode->pRaftStore);
+  raftStoreClearVote(pSyncNode->pRaftStore);
+  voteGrantedReset(pSyncNode->pVotesGranted, pSyncNode->pRaftStore->currentTerm);
+  votesRespondReset(pSyncNode->pVotesRespond, pSyncNode->pRaftStore->currentTerm);
+
+  syncNodeVoteForSelf(pSyncNode);
   int32_t ret = syncNodeRequestVotePeers(pSyncNode);
+  assert(ret == 0);
+  syncNodeResetElectTimer(pSyncNode);
+
   return ret;
 }
 

source/libs/sync/src/syncEnv.c

@@ -19,19 +19,18 @@
 SSyncEnv *gSyncEnv = NULL;
 
 // local function -----------------
-static void    syncEnvTick(void *param, void *tmrId);
-static int32_t doSyncEnvStart(SSyncEnv *pSyncEnv);
-static int32_t doSyncEnvStop(SSyncEnv *pSyncEnv);
-static tmr_h   doSyncEnvStartTimer(SSyncEnv *pSyncEnv, TAOS_TMR_CALLBACK fp, int mseconds, void *param);
-static void    doSyncEnvStopTimer(SSyncEnv *pSyncEnv, tmr_h *pTimer);
+static SSyncEnv *doSyncEnvStart();
+static int32_t   doSyncEnvStop(SSyncEnv *pSyncEnv);
+static int32_t   doSyncEnvStartTimer(SSyncEnv *pSyncEnv);
+static int32_t   doSyncEnvStopTimer(SSyncEnv *pSyncEnv);
+static void      syncEnvTick(void *param, void *tmrId);
 // --------------------------------
 
 int32_t syncEnvStart() {
-  int32_t ret;
+  int32_t ret = 0;
   taosSeedRand(taosGetTimestampSec());
-  gSyncEnv = (SSyncEnv *)malloc(sizeof(SSyncEnv));
+  gSyncEnv = doSyncEnvStart(gSyncEnv);
   assert(gSyncEnv != NULL);
-  ret = doSyncEnvStart(gSyncEnv);
   return ret;
 }
 
@@ -40,31 +39,52 @@ int32_t syncEnvStop() {
   return ret;
 }
 
-tmr_h syncEnvStartTimer(TAOS_TMR_CALLBACK fp, int mseconds, void *param) {
-  return doSyncEnvStartTimer(gSyncEnv, fp, mseconds, param);
+int32_t syncEnvStartTimer() {
+  int32_t ret = doSyncEnvStartTimer(gSyncEnv);
+  return ret;
 }
 
-void syncEnvStopTimer(tmr_h *pTimer) { doSyncEnvStopTimer(gSyncEnv, pTimer); }
+int32_t syncEnvStopTimer() {
+  int32_t ret = doSyncEnvStopTimer(gSyncEnv);
+  return ret;
+}
 
 // local function -----------------
 static void syncEnvTick(void *param, void *tmrId) {
   SSyncEnv *pSyncEnv = (SSyncEnv *)param;
-  sTrace("syncEnvTick ... name:%s ", pSyncEnv->name);
-
-  pSyncEnv->pEnvTickTimer = taosTmrStart(syncEnvTick, 1000, pSyncEnv, pSyncEnv->pTimerManager);
+  if (atomic_load_64(&pSyncEnv->envTickTimerLogicClockUser) <= atomic_load_64(&pSyncEnv->envTickTimerLogicClock)) {
+    ++(pSyncEnv->envTickTimerCounter);
+    sTrace(
+        "syncEnvTick do ... envTickTimerLogicClockUser:%lu, envTickTimerLogicClock:%lu, envTickTimerCounter:%lu, "
+        "envTickTimerMS:%d, tmrId:%p",
+        pSyncEnv->envTickTimerLogicClockUser, pSyncEnv->envTickTimerLogicClock, pSyncEnv->envTickTimerCounter,
+        pSyncEnv->envTickTimerMS, tmrId);
+
+    // do something, tick ...
+    taosTmrReset(syncEnvTick, pSyncEnv->envTickTimerMS, pSyncEnv, pSyncEnv->pTimerManager, &pSyncEnv->pEnvTickTimer);
+  } else {
+    sTrace(
+        "syncEnvTick pass ... envTickTimerLogicClockUser:%lu, envTickTimerLogicClock:%lu, envTickTimerCounter:%lu, "
+        "envTickTimerMS:%d, tmrId:%p",
+        pSyncEnv->envTickTimerLogicClockUser, pSyncEnv->envTickTimerLogicClock, pSyncEnv->envTickTimerCounter,
+        pSyncEnv->envTickTimerMS, tmrId);
+  }
 }
 
-static int32_t doSyncEnvStart(SSyncEnv *pSyncEnv) {
-  snprintf(pSyncEnv->name, sizeof(pSyncEnv->name), "SyncEnv_%p", pSyncEnv);
+static SSyncEnv *doSyncEnvStart() {
+  SSyncEnv *pSyncEnv = (SSyncEnv *)malloc(sizeof(SSyncEnv));
+  assert(pSyncEnv != NULL);
+  memset(pSyncEnv, 0, sizeof(pSyncEnv));
+
+  pSyncEnv->envTickTimerCounter = 0;
+  pSyncEnv->envTickTimerMS = ENV_TICK_TIMER_MS;
+  pSyncEnv->FpEnvTickTimer = syncEnvTick;
+  atomic_store_64(&pSyncEnv->envTickTimerLogicClock, 0);
+  atomic_store_64(&pSyncEnv->envTickTimerLogicClockUser, 0);
 
   // start tmr thread
   pSyncEnv->pTimerManager = taosTmrInit(1000, 50, 10000, "SYNC-ENV");
-
-  // pSyncEnv->pEnvTickTimer = taosTmrStart(syncEnvTick, 1000, pSyncEnv, pSyncEnv->pTimerManager);
-
-  sTrace("SyncEnv start ok, name:%s", pSyncEnv->name);
-
-  return 0;
+  return pSyncEnv;
 }
 
 static int32_t doSyncEnvStop(SSyncEnv *pSyncEnv) {
@@ -72,8 +92,18 @@ static int32_t doSyncEnvStop(SSyncEnv *pSyncEnv) {
   return 0;
 }
 
-static tmr_h doSyncEnvStartTimer(SSyncEnv *pSyncEnv, TAOS_TMR_CALLBACK fp, int mseconds, void *param) {
-  return taosTmrStart(fp, mseconds, pSyncEnv, pSyncEnv->pTimerManager);
+static int32_t doSyncEnvStartTimer(SSyncEnv *pSyncEnv) {
+  int32_t ret = 0;
+  taosTmrReset(pSyncEnv->FpEnvTickTimer, pSyncEnv->envTickTimerMS, pSyncEnv, pSyncEnv->pTimerManager,
+               &pSyncEnv->pEnvTickTimer);
+  atomic_store_64(&pSyncEnv->envTickTimerLogicClock, pSyncEnv->envTickTimerLogicClockUser);
+  return ret;
 }
 
-static void doSyncEnvStopTimer(SSyncEnv *pSyncEnv, tmr_h *pTimer) {}
+static int32_t doSyncEnvStopTimer(SSyncEnv *pSyncEnv) {
+  int32_t ret = 0;
+  atomic_add_fetch_64(&pSyncEnv->envTickTimerLogicClockUser, 1);
+  taosTmrStop(pSyncEnv->pEnvTickTimer);
+  pSyncEnv->pEnvTickTimer = NULL;
+  return ret;
+}

source/libs/sync/src/syncIO.c

@@ -16,6 +16,7 @@
 #include "syncIO.h"
 #include <tdatablock.h>
 #include "syncMessage.h"
+#include "syncUtil.h"
 #include "tglobal.h"
 #include "ttimer.h"
 #include "tutil.h"
@@ -23,33 +24,36 @@
 SSyncIO *gSyncIO = NULL;
 
 // local function ------------
-static int32_t  syncIOStartInternal(SSyncIO *io);
-static int32_t  syncIOStopInternal(SSyncIO *io);
 static SSyncIO *syncIOCreate(char *host, uint16_t port);
 static int32_t  syncIODestroy(SSyncIO *io);
+static int32_t  syncIOStartInternal(SSyncIO *io);
+static int32_t  syncIOStopInternal(SSyncIO *io);
 
-static void *syncIOConsumerFunc(void *param);
-static int   syncIOAuth(void *parent, char *meterId, char *spi, char *encrypt, char *secret, char *ckey);
-static void  syncIOProcessRequest(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet);
-static void  syncIOProcessReply(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet);
-
-static int32_t syncIOTickQInternal(SSyncIO *io);
-static void    syncIOTickQFunc(void *param, void *tmrId);
-static int32_t syncIOTickPingInternal(SSyncIO *io);
-static void    syncIOTickPingFunc(void *param, void *tmrId);
+static void   *syncIOConsumerFunc(void *param);
+static void    syncIOProcessRequest(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet);
+static void    syncIOProcessReply(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet);
+static int32_t syncIOAuth(void *parent, char *meterId, char *spi, char *encrypt, char *secret, char *ckey);
+
+static int32_t syncIOStartQ(SSyncIO *io);
+static int32_t syncIOStopQ(SSyncIO *io);
+static int32_t syncIOStartPing(SSyncIO *io);
+static int32_t syncIOStopPing(SSyncIO *io);
+static void    syncIOTickQ(void *param, void *tmrId);
+static void    syncIOTickPing(void *param, void *tmrId);
 // ----------------------------
 
 // public function ------------
 int32_t syncIOStart(char *host, uint16_t port) {
+  int32_t ret = 0;
   gSyncIO = syncIOCreate(host, port);
   assert(gSyncIO != NULL);
 
   taosSeedRand(taosGetTimestampSec());
-  int32_t ret = syncIOStartInternal(gSyncIO);
+  ret = syncIOStartInternal(gSyncIO);
   assert(ret == 0);
 
-  sTrace("syncIOStart ok, gSyncIO:%p gSyncIO->clientRpc:%p", gSyncIO, gSyncIO->clientRpc);
-  return 0;
+  sTrace("syncIOStart ok, gSyncIO:%p", gSyncIO);
+  return ret;
 }
 
 int32_t syncIOStop() {
@@ -61,37 +65,25 @@ int32_t syncIOStop() {
   return ret;
 }
 
-int32_t syncIOTickQ() {
-  int32_t ret = syncIOTickQInternal(gSyncIO);
-  assert(ret == 0);
-  return ret;
-}
+int32_t syncIOSendMsg(void *clientRpc, const SEpSet *pEpSet, SRpcMsg *pMsg) {
+  assert(pEpSet->inUse == 0);
+  assert(pEpSet->numOfEps == 1);
 
-int32_t syncIOTickPing() {
-  int32_t ret = syncIOTickPingInternal(gSyncIO);
-  assert(ret == 0);
-  return ret;
-}
+  int32_t ret = 0;
+  char    logBuf[256];
+  snprintf(logBuf, sizeof(logBuf), "==syncIOSendMsg== %s:%d", pEpSet->eps[0].fqdn, pEpSet->eps[0].port);
+  syncRpcMsgPrint2(logBuf, pMsg);
 
-int32_t syncIOSendMsg(void *clientRpc, const SEpSet *pEpSet, SRpcMsg *pMsg) {
-  sTrace(
-      "<--- syncIOSendMsg ---> clientRpc:%p, numOfEps:%d, inUse:%d, destAddr:%s-%u, pMsg->ahandle:%p, pMsg->handle:%p, "
-      "pMsg->msgType:%d, pMsg->contLen:%d",
-      clientRpc, pEpSet->numOfEps, pEpSet->inUse, pEpSet->eps[0].fqdn, pEpSet->eps[0].port, pMsg->ahandle, pMsg->handle,
-      pMsg->msgType, pMsg->contLen);
-  {
-    cJSON *pJson = syncRpcMsg2Json(pMsg);
-    char * serialized = cJSON_Print(pJson);
-    sTrace("process syncMessage send: pMsg:%s ", serialized);
-    free(serialized);
-    cJSON_Delete(pJson);
-  }
   pMsg->handle = NULL;
   rpcSendRequest(clientRpc, pEpSet, pMsg, NULL);
-  return 0;
+  return ret;
 }
 
 int32_t syncIOEqMsg(void *queue, SRpcMsg *pMsg) {
+  int32_t ret = 0;
+  char    logBuf[128];
+  syncRpcMsgPrint2((char *)"==syncIOEqMsg==", pMsg);
+
   SRpcMsg *pTemp;
   pTemp = taosAllocateQitem(sizeof(SRpcMsg));
   memcpy(pTemp, pMsg, sizeof(SRpcMsg));
@@ -99,11 +91,75 @@ int32_t syncIOEqMsg(void *queue, SRpcMsg *pMsg) {
   STaosQueue *pMsgQ = queue;
   taosWriteQitem(pMsgQ, pTemp);
 
-  return 0;
+  return ret;
+}
+
+int32_t syncIOQTimerStart() {
+  int32_t ret = syncIOStartQ(gSyncIO);
+  assert(ret == 0);
+  return ret;
+}
+
+int32_t syncIOQTimerStop() {
+  int32_t ret = syncIOStopQ(gSyncIO);
+  assert(ret == 0);
+  return ret;
+}
+
+int32_t syncIOPingTimerStart() {
+  int32_t ret = syncIOStartPing(gSyncIO);
+  assert(ret == 0);
+  return ret;
+}
+
+int32_t syncIOPingTimerStop() {
+  int32_t ret = syncIOStopPing(gSyncIO);
+  assert(ret == 0);
+  return ret;
 }
 
 // local function ------------
+static SSyncIO *syncIOCreate(char *host, uint16_t port) {
+  SSyncIO *io = (SSyncIO *)malloc(sizeof(SSyncIO));
+  memset(io, 0, sizeof(*io));
+
+  io->pMsgQ = taosOpenQueue();
+  io->pQset = taosOpenQset();
+  taosAddIntoQset(io->pQset, io->pMsgQ, NULL);
+
+  io->myAddr.inUse = 0;
+  io->myAddr.numOfEps = 0;
+  addEpIntoEpSet(&io->myAddr, host, port);
+
+  io->qTimerMS = TICK_Q_TIMER_MS;
+  io->pingTimerMS = TICK_Ping_TIMER_MS;
+
+  return io;
+}
+
+static int32_t syncIODestroy(SSyncIO *io) {
+  int32_t ret = 0;
+  int8_t  start = atomic_load_8(&io->isStart);
+  assert(start == 0);
+
+  if (io->serverRpc != NULL) {
+    rpcClose(io->serverRpc);
+    io->serverRpc = NULL;
+  }
+
+  if (io->clientRpc != NULL) {
+    rpcClose(io->clientRpc);
+    io->clientRpc = NULL;
+  }
+
+  taosCloseQueue(io->pMsgQ);
+  taosCloseQset(io->pQset);
+
+  return ret;
+}
+
 static int32_t syncIOStartInternal(SSyncIO *io) {
+  int32_t ret = 0;
   taosBlockSIGPIPE();
 
   rpcInit();
@@ -163,58 +219,24 @@ static int32_t syncIOStartInternal(SSyncIO *io) {
   }
 
   // start tmr thread
-  io->ioTimerManager = taosTmrInit(1000, 50, 10000, "SYNC");
+  io->timerMgr = taosTmrInit(1000, 50, 10000, "SYNC-IO");
 
-  return 0;
+  atomic_store_8(&io->isStart, 1);
+  return ret;
 }
 
 static int32_t syncIOStopInternal(SSyncIO *io) {
+  int32_t ret = 0;
   atomic_store_8(&io->isStart, 0);
   pthread_join(io->consumerTid, NULL);
-  return 0;
-}
-
-static SSyncIO *syncIOCreate(char *host, uint16_t port) {
-  SSyncIO *io = (SSyncIO *)malloc(sizeof(SSyncIO));
-  memset(io, 0, sizeof(*io));
-
-  io->pMsgQ = taosOpenQueue();
-  io->pQset = taosOpenQset();
-  taosAddIntoQset(io->pQset, io->pMsgQ, NULL);
-
-  io->myAddr.inUse = 0;
-  addEpIntoEpSet(&io->myAddr, host, port);
-
-  return io;
-}
-
-static int32_t syncIODestroy(SSyncIO *io) {
-  int8_t start = atomic_load_8(&io->isStart);
-  assert(start == 0);
-
-  if (io->serverRpc != NULL) {
-    free(io->serverRpc);
-    io->serverRpc = NULL;
-  }
-
-  if (io->clientRpc != NULL) {
-    free(io->clientRpc);
-    io->clientRpc = NULL;
-  }
-
-  taosCloseQueue(io->pMsgQ);
-  taosCloseQset(io->pQset);
-
-  return 0;
+  taosTmrCleanUp(io->timerMgr);
+  return ret;
 }
 
 static void *syncIOConsumerFunc(void *param) {
-  SSyncIO *io = param;
-
+  SSyncIO   *io = param;
   STaosQall *qall;
-  SRpcMsg *  pRpcMsg, rpcMsg;
-  int        type;
-
+  SRpcMsg   *pRpcMsg, rpcMsg;
   qall = taosAllocateQall();
 
   while (1) {
@@ -226,77 +248,74 @@ static void *syncIOConsumerFunc(void *param) {
 
     for (int i = 0; i < numOfMsgs; ++i) {
       taosGetQitem(qall, (void **)&pRpcMsg);
+      syncRpcMsgLog2((char *)"==syncIOConsumerFunc==", pRpcMsg);
 
-      char *s = syncRpcMsg2Str(pRpcMsg);
-      sTrace("syncIOConsumerFunc get item from queue: msgType:%d contLen:%d msg:%s", pRpcMsg->msgType, pRpcMsg->contLen,
-             s);
-      free(s);
-
+      // use switch case instead of if else
       if (pRpcMsg->msgType == SYNC_PING) {
         if (io->FpOnSyncPing != NULL) {
-          SyncPing *pSyncMsg;
+          SyncPing *pSyncMsg = syncPingFromRpcMsg2(pRpcMsg);
+          assert(pSyncMsg != NULL);
+          io->FpOnSyncPing(io->pSyncNode, pSyncMsg);
+          syncPingDestroy(pSyncMsg);
+          /*
           pSyncMsg = syncPingBuild(pRpcMsg->contLen);
           syncPingFromRpcMsg(pRpcMsg, pSyncMsg);
           // memcpy(pSyncMsg, tmpRpcMsg.pCont, tmpRpcMsg.contLen);
           io->FpOnSyncPing(io->pSyncNode, pSyncMsg);
           syncPingDestroy(pSyncMsg);
+          */
         }
 
       } else if (pRpcMsg->msgType == SYNC_PING_REPLY) {
         if (io->FpOnSyncPingReply != NULL) {
-          SyncPingReply *pSyncMsg;
-          pSyncMsg = syncPingReplyBuild(pRpcMsg->contLen);
-          syncPingReplyFromRpcMsg(pRpcMsg, pSyncMsg);
+          SyncPingReply *pSyncMsg = syncPingReplyFromRpcMsg2(pRpcMsg);
           io->FpOnSyncPingReply(io->pSyncNode, pSyncMsg);
           syncPingReplyDestroy(pSyncMsg);
         }
 
+      } else if (pRpcMsg->msgType == SYNC_CLIENT_REQUEST) {
+        if (io->FpOnSyncClientRequest != NULL) {
+          SyncClientRequest *pSyncMsg = syncClientRequestFromRpcMsg2(pRpcMsg);
+          io->FpOnSyncClientRequest(io->pSyncNode, pSyncMsg);
+          syncClientRequestDestroy(pSyncMsg);
+        }
+
       } else if (pRpcMsg->msgType == SYNC_REQUEST_VOTE) {
         if (io->FpOnSyncRequestVote != NULL) {
-          SyncRequestVote *pSyncMsg;
-          pSyncMsg = syncRequestVoteBuild(pRpcMsg->contLen);
-          syncRequestVoteFromRpcMsg(pRpcMsg, pSyncMsg);
+          SyncRequestVote *pSyncMsg = syncRequestVoteFromRpcMsg2(pRpcMsg);
           io->FpOnSyncRequestVote(io->pSyncNode, pSyncMsg);
           syncRequestVoteDestroy(pSyncMsg);
         }
 
       } else if (pRpcMsg->msgType == SYNC_REQUEST_VOTE_REPLY) {
         if (io->FpOnSyncRequestVoteReply != NULL) {
-          SyncRequestVoteReply *pSyncMsg;
-          pSyncMsg = syncRequestVoteReplyBuild();
-          syncRequestVoteReplyFromRpcMsg(pRpcMsg, pSyncMsg);
+          SyncRequestVoteReply *pSyncMsg = syncRequestVoteReplyFromRpcMsg2(pRpcMsg);
           io->FpOnSyncRequestVoteReply(io->pSyncNode, pSyncMsg);
           syncRequestVoteReplyDestroy(pSyncMsg);
         }
 
       } else if (pRpcMsg->msgType == SYNC_APPEND_ENTRIES) {
         if (io->FpOnSyncAppendEntries != NULL) {
-          SyncAppendEntries *pSyncMsg;
-          pSyncMsg = syncAppendEntriesBuild(pRpcMsg->contLen);
-          syncAppendEntriesFromRpcMsg(pRpcMsg, pSyncMsg);
+          SyncAppendEntries *pSyncMsg = syncAppendEntriesFromRpcMsg2(pRpcMsg);
           io->FpOnSyncAppendEntries(io->pSyncNode, pSyncMsg);
           syncAppendEntriesDestroy(pSyncMsg);
         }
 
       } else if (pRpcMsg->msgType == SYNC_APPEND_ENTRIES_REPLY) {
         if (io->FpOnSyncAppendEntriesReply != NULL) {
-          SyncAppendEntriesReply *pSyncMsg;
-          pSyncMsg = syncAppendEntriesReplyBuild();
-          syncAppendEntriesReplyFromRpcMsg(pRpcMsg, pSyncMsg);
+          SyncAppendEntriesReply *pSyncMsg = syncAppendEntriesReplyFromRpcMsg2(pRpcMsg);
           io->FpOnSyncAppendEntriesReply(io->pSyncNode, pSyncMsg);
           syncAppendEntriesReplyDestroy(pSyncMsg);
         }
 
       } else if (pRpcMsg->msgType == SYNC_TIMEOUT) {
         if (io->FpOnSyncTimeout != NULL) {
-          SyncTimeout *pSyncMsg;
-          pSyncMsg = syncTimeoutBuild();
-          syncTimeoutFromRpcMsg(pRpcMsg, pSyncMsg);
+          SyncTimeout *pSyncMsg = syncTimeoutFromRpcMsg2(pRpcMsg);
           io->FpOnSyncTimeout(io->pSyncNode, pSyncMsg);
           syncTimeoutDestroy(pSyncMsg);
         }
       } else {
-        ;
+        sTrace("unknown msgType:%d, no operator", pRpcMsg->msgType);
       }
     }
 
@@ -306,15 +325,16 @@ static void *syncIOConsumerFunc(void *param) {
       rpcFreeCont(pRpcMsg->pCont);
 
       if (pRpcMsg->handle != NULL) {
-        int msgSize = 128;
+        int msgSize = 32;
         memset(&rpcMsg, 0, sizeof(rpcMsg));
+        rpcMsg.msgType = SYNC_RESPONSE;
         rpcMsg.pCont = rpcMallocCont(msgSize);
         rpcMsg.contLen = msgSize;
         snprintf(rpcMsg.pCont, rpcMsg.contLen, "%s", "give a reply");
         rpcMsg.handle = pRpcMsg->handle;
         rpcMsg.code = 0;
 
-        sTrace("syncIOConsumerFunc rpcSendResponse ... msgType:%d contLen:%d", pRpcMsg->msgType, rpcMsg.contLen);
+        syncRpcMsgPrint2((char *)"syncIOConsumerFunc rpcSendResponse --> ", &rpcMsg);
         rpcSendResponse(&rpcMsg);
       }
 
@@ -326,71 +346,95 @@ static void *syncIOConsumerFunc(void *param) {
   return NULL;
 }
 
-static int syncIOAuth(void *parent, char *meterId, char *spi, char *encrypt, char *secret, char *ckey) {
-  // app shall retrieve the auth info based on meterID from DB or a data file
-  // demo code here only for simple demo
-  int ret = 0;
-  return ret;
-}
-
 static void syncIOProcessRequest(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet) {
-  sTrace("<-- syncIOProcessRequest --> type:%d, contLen:%d, cont:%s", pMsg->msgType, pMsg->contLen,
-         (char *)pMsg->pCont);
-
+  syncRpcMsgPrint2((char *)"==syncIOProcessRequest==", pMsg);
   SSyncIO *io = pParent;
   SRpcMsg *pTemp;
-
   pTemp = taosAllocateQitem(sizeof(SRpcMsg));
   memcpy(pTemp, pMsg, sizeof(SRpcMsg));
-
   taosWriteQitem(io->pMsgQ, pTemp);
 }
 
 static void syncIOProcessReply(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet) {
-  sTrace("syncIOProcessReply: type:%d, contLen:%d msg:%s", pMsg->msgType, pMsg->contLen, (char *)pMsg->pCont);
+  if (pMsg->msgType == SYNC_RESPONSE) {
+    sTrace("==syncIOProcessReply==");
+  } else {
+    syncRpcMsgPrint2((char *)"==syncIOProcessReply==", pMsg);
+  }
   rpcFreeCont(pMsg->pCont);
 }
 
-static int32_t syncIOTickQInternal(SSyncIO *io) {
-  io->ioTimerTickQ = taosTmrStart(syncIOTickQFunc, 1000, io, io->ioTimerManager);
-  return 0;
+static int32_t syncIOAuth(void *parent, char *meterId, char *spi, char *encrypt, char *secret, char *ckey) {
+  // app shall retrieve the auth info based on meterID from DB or a data file
+  // demo code here only for simple demo
+  int32_t ret = 0;
+  return ret;
+}
+
+static int32_t syncIOStartQ(SSyncIO *io) {
+  int32_t ret = 0;
+  taosTmrReset(syncIOTickQ, io->qTimerMS, io, io->timerMgr, &io->qTimer);
+  return ret;
 }
 
-static void syncIOTickQFunc(void *param, void *tmrId) {
+static int32_t syncIOStopQ(SSyncIO *io) {
+  int32_t ret = 0;
+  taosTmrStop(io->qTimer);
+  io->qTimer = NULL;
+  return ret;
+}
+
+static int32_t syncIOStartPing(SSyncIO *io) {
+  int32_t ret = 0;
+  taosTmrReset(syncIOTickPing, io->pingTimerMS, io, io->timerMgr, &io->pingTimer);
+  return ret;
+}
+
+static int32_t syncIOStopPing(SSyncIO *io) {
+  int32_t ret = 0;
+  taosTmrStop(io->pingTimer);
+  io->pingTimer = NULL;
+  return ret;
+}
+
+static void syncIOTickQ(void *param, void *tmrId) {
   SSyncIO *io = (SSyncIO *)param;
-  sTrace("<-- syncIOTickQFunc -->");
 
-  SRpcMsg rpcMsg;
-  rpcMsg.contLen = 64;
-  rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
-  snprintf(rpcMsg.pCont, rpcMsg.contLen, "%s", "syncIOTickQ");
-  rpcMsg.handle = NULL;
-  rpcMsg.msgType = 55;
+  SRaftId srcId, destId;
+  srcId.addr = syncUtilAddr2U64(io->myAddr.eps[0].fqdn, io->myAddr.eps[0].port);
+  srcId.vgId = -1;
+  destId.addr = syncUtilAddr2U64(io->myAddr.eps[0].fqdn, io->myAddr.eps[0].port);
+  destId.vgId = -1;
+  SyncPingReply *pMsg = syncPingReplyBuild2(&srcId, &destId, "syncIOTickQ");
 
+  SRpcMsg rpcMsg;
+  syncPingReply2RpcMsg(pMsg, &rpcMsg);
   SRpcMsg *pTemp;
   pTemp = taosAllocateQitem(sizeof(SRpcMsg));
   memcpy(pTemp, &rpcMsg, sizeof(SRpcMsg));
-
+  syncRpcMsgPrint2((char *)"==syncIOTickQ==", &rpcMsg);
   taosWriteQitem(io->pMsgQ, pTemp);
-  taosTmrReset(syncIOTickQFunc, 1000, io, io->ioTimerManager, &io->ioTimerTickQ);
-}
+  syncPingReplyDestroy(pMsg);
 
-static int32_t syncIOTickPingInternal(SSyncIO *io) {
-  io->ioTimerTickPing = taosTmrStart(syncIOTickPingFunc, 1000, io, io->ioTimerManager);
-  return 0;
+  taosTmrReset(syncIOTickQ, io->qTimerMS, io, io->timerMgr, &io->qTimer);
 }
 
-static void syncIOTickPingFunc(void *param, void *tmrId) {
+static void syncIOTickPing(void *param, void *tmrId) {
   SSyncIO *io = (SSyncIO *)param;
-  sTrace("<-- syncIOTickPingFunc -->");
 
-  SRpcMsg rpcMsg;
-  rpcMsg.contLen = 64;
-  rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
-  snprintf(rpcMsg.pCont, rpcMsg.contLen, "%s", "syncIOTickPing");
-  rpcMsg.handle = NULL;
-  rpcMsg.msgType = 77;
+  SRaftId srcId, destId;
+  srcId.addr = syncUtilAddr2U64(io->myAddr.eps[0].fqdn, io->myAddr.eps[0].port);
+  srcId.vgId = -1;
+  destId.addr = syncUtilAddr2U64(io->myAddr.eps[0].fqdn, io->myAddr.eps[0].port);
+  destId.vgId = -1;
+  SyncPing *pMsg = syncPingBuild2(&srcId, &destId, "syncIOTickPing");
+  // SyncPing *pMsg = syncPingBuild3(&srcId, &destId);
 
+  SRpcMsg rpcMsg;
+  syncPing2RpcMsg(pMsg, &rpcMsg);
+  syncRpcMsgPrint2((char *)"==syncIOTickPing==", &rpcMsg);
   rpcSendRequest(io->clientRpc, &io->myAddr, &rpcMsg, NULL);
-  taosTmrReset(syncIOTickPingFunc, 1000, io, io->ioTimerManager, &io->ioTimerTickPing);
+  syncPingDestroy(pMsg);
+
+  taosTmrReset(syncIOTickPing, io->pingTimerMS, io, io->timerMgr, &io->pingTimer);
 }
\ No newline at end of file

source/libs/sync/src/syncMessage.c

@@ -65,10 +65,32 @@ cJSON* syncRpcMsg2Json(SRpcMsg* pRpcMsg) {
     pRoot = syncAppendEntriesReply2Json(pSyncMsg);
     syncAppendEntriesReplyDestroy(pSyncMsg);
 
+  } else if (pRpcMsg->msgType == SYNC_RESPONSE) {
+    pRoot = cJSON_CreateObject();
+    char* s;
+    s = syncUtilprintBin((char*)(pRpcMsg->pCont), pRpcMsg->contLen);
+    cJSON_AddStringToObject(pRoot, "pCont", s);
+    free(s);
+    s = syncUtilprintBin2((char*)(pRpcMsg->pCont), pRpcMsg->contLen);
+    cJSON_AddStringToObject(pRoot, "pCont2", s);
+    free(s);
+
   } else {
     pRoot = syncRpcUnknownMsg2Json();
+    char* s;
+    s = syncUtilprintBin((char*)(pRpcMsg->pCont), pRpcMsg->contLen);
+    cJSON_AddStringToObject(pRoot, "pCont", s);
+    free(s);
+    s = syncUtilprintBin2((char*)(pRpcMsg->pCont), pRpcMsg->contLen);
+    cJSON_AddStringToObject(pRoot, "pCont2", s);
+    free(s);
   }
 
+  cJSON_AddNumberToObject(pRoot, "msgType", pRpcMsg->msgType);
+  cJSON_AddNumberToObject(pRoot, "contLen", pRpcMsg->contLen);
+  cJSON_AddNumberToObject(pRoot, "code", pRpcMsg->code);
+  // cJSON_AddNumberToObject(pRoot, "persist", pRpcMsg->persist);
+
   cJSON* pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "RpcMsg", pRoot);
   return pJson;
@@ -77,7 +99,7 @@ cJSON* syncRpcMsg2Json(SRpcMsg* pRpcMsg) {
 cJSON* syncRpcUnknownMsg2Json() {
   cJSON* pRoot = cJSON_CreateObject();
   cJSON_AddNumberToObject(pRoot, "msgType", SYNC_UNKNOWN);
-  cJSON_AddStringToObject(pRoot, "data", "known message");
+  cJSON_AddStringToObject(pRoot, "data", "unknown message");
 
   cJSON* pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SyncUnknown", pRoot);
@@ -798,8 +820,8 @@ cJSON* syncRequestVote2Json(const SyncRequestVote* pMsg) {
   cJSON_AddNumberToObject(pDestId, "vgId", pMsg->destId.vgId);
   cJSON_AddItemToObject(pRoot, "destId", pDestId);
 
-  snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->currentTerm);
-  cJSON_AddStringToObject(pRoot, "currentTerm", u64buf);
+  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);
@@ -1086,6 +1108,9 @@ cJSON* syncAppendEntries2Json(const SyncAppendEntries* pMsg) {
   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);
 
@@ -1242,9 +1267,11 @@ cJSON* syncAppendEntriesReply2Json(const SyncAppendEntriesReply* pMsg) {
   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, "match_index", u64buf);
+  cJSON_AddStringToObject(pRoot, "matchIndex", u64buf);
 
   cJSON* pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SyncAppendEntriesReply", pRoot);
@@ -1283,4 +1310,4 @@ void syncAppendEntriesReplyLog2(char* s, const SyncAppendEntriesReply* pMsg) {
   char* serialized = syncAppendEntriesReply2Str(pMsg);
   sTrace("syncAppendEntriesReplyLog2 | len:%lu | %s | %s", strlen(serialized), s, serialized);
   free(serialized);
-}
\ No newline at end of file
+}

source/libs/sync/src/syncRaftLog.c

@@ -130,7 +130,7 @@ cJSON* logStore2Json(SSyncLogStore* pLogStore) {
   cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf);
   snprintf(u64buf, sizeof(u64buf), "%p", pData->pWal);
   cJSON_AddStringToObject(pRoot, "pWal", u64buf);
-  snprintf(u64buf, sizeof(u64buf), "%lu", logStoreLastIndex(pLogStore));
+  snprintf(u64buf, sizeof(u64buf), "%ld", logStoreLastIndex(pLogStore));
   cJSON_AddStringToObject(pRoot, "LastIndex", u64buf);
   snprintf(u64buf, sizeof(u64buf), "%lu", logStoreLastTerm(pLogStore));
   cJSON_AddStringToObject(pRoot, "LastTerm", u64buf);

source/libs/sync/src/syncRaftStore.c

@@ -15,6 +15,8 @@
 
 #include "syncRaftStore.h"
 #include "cJSON.h"
+#include "syncEnv.h"
+#include "syncUtil.h"
 
 // private function
 static int32_t raftStoreInit(SRaftStore *pRaftStore);
@@ -135,6 +137,33 @@ int32_t raftStoreDeserialize(SRaftStore *pRaftStore, char *buf, size_t len) {
   return 0;
 }
 
+bool raftStoreHasVoted(SRaftStore *pRaftStore) {
+  bool b = syncUtilEmptyId(&(pRaftStore->voteFor));
+  return b;
+}
+
+void raftStoreVote(SRaftStore *pRaftStore, SRaftId *pRaftId) {
+  assert(!raftStoreHasVoted(pRaftStore));
+  assert(!syncUtilEmptyId(pRaftId));
+  pRaftStore->voteFor = *pRaftId;
+  raftStorePersist(pRaftStore);
+}
+
+void raftStoreClearVote(SRaftStore *pRaftStore) {
+  pRaftStore->voteFor = EMPTY_RAFT_ID;
+  raftStorePersist(pRaftStore);
+}
+
+void raftStoreNextTerm(SRaftStore *pRaftStore) {
+  ++(pRaftStore->currentTerm);
+  raftStorePersist(pRaftStore);
+}
+
+void raftStoreSetTerm(SRaftStore *pRaftStore, SyncTerm term) {
+  pRaftStore->currentTerm = term;
+  raftStorePersist(pRaftStore);
+}
+
 // for debug -------------------
 void raftStorePrint(SRaftStore *pObj) {
   char serialized[RAFT_STORE_BLOCK_SIZE];

source/libs/sync/src/syncReplication.c

@@ -14,7 +14,11 @@
  */
 
 #include "syncReplication.h"
+#include "syncIndexMgr.h"
 #include "syncMessage.h"
+#include "syncRaftEntry.h"
+#include "syncRaftLog.h"
+#include "syncUtil.h"
 
 // TLA+ Spec
 // AppendEntries(i, j) ==
@@ -42,7 +46,39 @@
 //    /\ UNCHANGED <<serverVars, candidateVars, leaderVars, logVars>>
 //
 int32_t syncNodeAppendEntriesPeers(SSyncNode* pSyncNode) {
+  assert(pSyncNode->state == TAOS_SYNC_STATE_LEADER);
+
   int32_t ret = 0;
+  for (int i = 0; i < pSyncNode->peersNum; ++i) {
+    SRaftId*  pDestId = &(pSyncNode->peersId[i]);
+    SyncIndex nextIndex = syncIndexMgrGetIndex(pSyncNode->pNextIndex, pDestId);
+
+    SyncIndex preLogIndex = nextIndex - 1;
+
+    SyncTerm preLogTerm = 0;
+    if (preLogIndex >= SYNC_INDEX_BEGIN) {
+      SSyncRaftEntry* pPreEntry = pSyncNode->pLogStore->getEntry(pSyncNode->pLogStore, preLogIndex);
+      preLogTerm = pPreEntry->term;
+    }
+
+    SyncIndex lastIndex = syncUtilMinIndex(pSyncNode->pLogStore->getLastIndex(pSyncNode->pLogStore), nextIndex);
+    assert(nextIndex == lastIndex);
+
+    SSyncRaftEntry* pEntry = logStoreGetEntry(pSyncNode->pLogStore, nextIndex);
+    assert(pEntry != NULL);
+
+    SyncAppendEntries* pMsg = syncAppendEntriesBuild(pEntry->bytes);
+    pMsg->srcId = pSyncNode->myRaftId;
+    pMsg->destId = *pDestId;
+    pMsg->prevLogIndex = preLogIndex;
+    pMsg->prevLogTerm = preLogTerm;
+    pMsg->commitIndex = pSyncNode->commitIndex;
+    pMsg->dataLen = pEntry->bytes;
+    // add pEntry into msg
+
+    syncNodeAppendEntries(pSyncNode, pDestId, pMsg);
+  }
+
   return ret;
 }
 

source/libs/sync/src/syncRequestVote.c

@@ -14,6 +14,10 @@
  */
 
 #include "syncRequestVote.h"
+#include "syncInt.h"
+#include "syncRaftStore.h"
+#include "syncUtil.h"
+#include "syncVoteMgr.h"
 
 // TLA+ Spec
 // HandleRequestVoteRequest(i, j, m) ==
@@ -37,4 +41,34 @@
 //                 m)
 //       /\ UNCHANGED <<state, currentTerm, candidateVars, leaderVars, logVars>>
 //
-int32_t syncNodeOnRequestVoteCb(SSyncNode* ths, SyncRequestVote* pMsg) {}
+int32_t syncNodeOnRequestVoteCb(SSyncNode* ths, SyncRequestVote* pMsg) {
+  int32_t ret = 0;
+  syncRequestVoteLog2("==syncNodeOnRequestVoteCb==", pMsg);
+
+  if (pMsg->term > ths->pRaftStore->currentTerm) {
+    syncNodeUpdateTerm(ths, pMsg->term);
+  }
+  assert(pMsg->term <= ths->pRaftStore->currentTerm);
+
+  bool logOK = (pMsg->lastLogTerm > ths->pLogStore->getLastTerm(ths->pLogStore)) ||
+               ((pMsg->lastLogTerm == ths->pLogStore->getLastTerm(ths->pLogStore)) &&
+                (pMsg->lastLogIndex >= ths->pLogStore->getLastIndex(ths->pLogStore)));
+  bool grant = (pMsg->term == ths->pRaftStore->currentTerm) && logOK &&
+               ((!raftStoreHasVoted(ths->pRaftStore)) || (syncUtilSameId(&(ths->pRaftStore->voteFor), &(pMsg->srcId))));
+  if (grant) {
+    raftStoreVote(ths->pRaftStore, &(pMsg->srcId));
+  }
+
+  SyncRequestVoteReply* pReply = syncRequestVoteReplyBuild();
+  pReply->srcId = ths->myRaftId;
+  pReply->destId = pMsg->srcId;
+  pReply->term = ths->pRaftStore->currentTerm;
+  pReply->voteGranted = grant;
+
+  SRpcMsg rpcMsg;
+  syncRequestVoteReply2RpcMsg(pReply, &rpcMsg);
+  syncNodeSendMsgById(&pReply->destId, ths, &rpcMsg);
+  syncRequestVoteReplyDestroy(pReply);
+
+  return ret;
+}

source/libs/sync/src/syncRequestVoteReply.c

@@ -14,6 +14,10 @@
  */
 
 #include "syncRequestVoteReply.h"
+#include "syncInt.h"
+#include "syncRaftStore.h"
+#include "syncUtil.h"
+#include "syncVoteMgr.h"
 
 // TLA+ Spec
 // HandleRequestVoteResponse(i, j, m) ==
@@ -32,4 +36,33 @@
 //    /\ Discard(m)
 //    /\ UNCHANGED <<serverVars, votedFor, leaderVars, logVars>>
 //
-int32_t syncNodeOnRequestVoteReplyCb(SSyncNode* ths, SyncRequestVoteReply* pMsg) {}
+int32_t syncNodeOnRequestVoteReplyCb(SSyncNode* ths, SyncRequestVoteReply* pMsg) {
+  int32_t ret = 0;
+  syncRequestVoteReplyLog2("==syncNodeOnRequestVoteReplyCb==", pMsg);
+
+  if (pMsg->term < ths->pRaftStore->currentTerm) {
+    sTrace("DropStaleResponse, receive term:%lu, current term:%lu", pMsg->term, ths->pRaftStore->currentTerm);
+    return ret;
+  }
+
+  // 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);
+  //  }
+
+  assert(pMsg->term == ths->pRaftStore->currentTerm);
+
+  if (ths->state == TAOS_SYNC_STATE_CANDIDATE) {
+    votesRespondAdd(ths->pVotesRespond, pMsg);
+    if (pMsg->voteGranted) {
+      voteGrantedVote(ths->pVotesGranted, pMsg);
+      if (voteGrantedMajority(ths->pVotesGranted)) {
+        if (ths->pVotesGranted->toLeader) {
+          syncNodeCandidate2Leader(ths);
+          ths->pVotesGranted->toLeader = true;
+        }
+      }
+    }
+  }
+  return ret;
+}

source/libs/sync/src/syncTimeout.c

@@ -19,15 +19,7 @@
 
 int32_t syncNodeOnTimeoutCb(SSyncNode* ths, SyncTimeout* pMsg) {
   int32_t ret = 0;
-  sTrace("<-- syncNodeOnTimeoutCb -->");
-
-  {
-    cJSON* pJson = syncTimeout2Json(pMsg);
-    char*  serialized = cJSON_Print(pJson);
-    sTrace("process syncMessage recv: syncNodeOnTimeoutCb pMsg:%s ", serialized);
-    free(serialized);
-    cJSON_Delete(pJson);
-  }
+  syncTimeoutLog2("==syncNodeOnTimeoutCb==", pMsg);
 
   if (pMsg->timeoutType == SYNC_TIMEOUT_PING) {
     if (atomic_load_64(&ths->pingTimerLogicClockUser) <= pMsg->logicClock) {

source/libs/sync/test/syncIOTickPingTest.cpp

@@ -25,8 +25,15 @@ int main() {
   ret = syncIOStart((char*)"127.0.0.1", 7010);
   assert(ret == 0);
 
-  ret = syncIOTickPing();
-  assert(ret == 0);
+  for (int i = 0; i < 3; ++i) {
+    ret = syncIOPingTimerStart();
+    assert(ret == 0);
+    taosMsleep(5000);
+
+    ret = syncIOPingTimerStop();
+    assert(ret == 0);
+    taosMsleep(5000);
+  }
 
   while (1) {
     taosSsleep(1);