refactor(sync): add resp ttl clean

include/libs/sync/sync.h

@@ -26,6 +26,8 @@ extern "C" {
 
 extern bool gRaftDetailLog;
 
+#define SYNC_RESP_TTL_MS 5000
+
 #define SYNC_MAX_BATCH_SIZE 500
 #define SYNC_INDEX_BEGIN 0
 #define SYNC_INDEX_INVALID -1

source/dnode/vnode/src/vnd/vnodeSync.c

@@ -427,13 +427,22 @@ static void vnodeSyncCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta c
          syncGetVgId(pVnode->sync), pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state,
          syncUtilState2String(cbMeta.state), pMsg->msgType, TMSG_INFO(pMsg->msgType));
 
-  SRpcMsg rpcMsg = {.msgType = pMsg->msgType, .contLen = pMsg->contLen};
-  rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
-  memcpy(rpcMsg.pCont, pMsg->pCont, pMsg->contLen);
-  syncGetAndDelRespRpc(pVnode->sync, cbMeta.seqNum, &rpcMsg.info);
-  rpcMsg.info.conn.applyIndex = cbMeta.index;
-  rpcMsg.info.conn.applyTerm = cbMeta.term;
-  tmsgPutToQueue(&pVnode->msgCb, APPLY_QUEUE, &rpcMsg);
+  if (cbMeta.code == 0) {
+    SRpcMsg rpcMsg = {.msgType = pMsg->msgType, .contLen = pMsg->contLen};
+    rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
+    memcpy(rpcMsg.pCont, pMsg->pCont, pMsg->contLen);
+    syncGetAndDelRespRpc(pVnode->sync, cbMeta.seqNum, &rpcMsg.info);
+    rpcMsg.info.conn.applyIndex = cbMeta.index;
+    rpcMsg.info.conn.applyTerm = cbMeta.term;
+    tmsgPutToQueue(&pVnode->msgCb, APPLY_QUEUE, &rpcMsg);
+  } else {
+    SRpcMsg rsp = {.code = cbMeta.code, .info = pMsg->info};
+    vError("vgId:%d, sync commit error, msgtype:%d,%s, error:0x%X, errmsg:%s", syncGetVgId(pVnode->sync), pMsg->msgType,
+           TMSG_INFO(pMsg->msgType), cbMeta.code, tstrerror(cbMeta.code));
+    if (rsp.info.handle != NULL) {
+      tmsgSendRsp(&rsp);
+    }
+  }
 }
 
 static void vnodeSyncPreCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) {

source/libs/sync/src/syncMain.c

@@ -1055,19 +1055,12 @@ SSyncNode* syncNodeOpen(const SSyncInfo* pOldSyncInfo) {
   }
 
   // tools
-  pSyncNode->pSyncRespMgr = syncRespMgrCreate(pSyncNode, 0);
+  pSyncNode->pSyncRespMgr = syncRespMgrCreate(pSyncNode, SYNC_RESP_TTL_MS);
   ASSERT(pSyncNode->pSyncRespMgr != NULL);
 
   // restore state
   pSyncNode->restoreFinish = false;
 
-  // pSyncNode->pSnapshot = NULL;
-  // if (pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
-  //   pSyncNode->pSnapshot = taosMemoryMalloc(sizeof(SSnapshot));
-  //   pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, pSyncNode->pSnapshot);
-  // }
-  // tsem_init(&(pSyncNode->restoreSem), 0, 0);
-
   // snapshot senders
   for (int i = 0; i < TSDB_MAX_REPLICA; ++i) {
     SSyncSnapshotSender* pSender = snapshotSenderCreate(pSyncNode, i);

source/libs/sync/src/syncRespMgr.c

@@ -122,54 +122,43 @@ void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl) {
   int        cnt = 0;
   SSyncNode *pSyncNode = pObj->data;
 
-  SArray *delIndexArray = taosArrayInit(0, sizeof(SyncIndex));
+  SArray *delIndexArray = taosArrayInit(0, sizeof(uint64_t));
   ASSERT(delIndexArray != NULL);
 
   while (pStub) {
-    size_t     len;
-    void *     key = taosHashGetKey(pStub, &len);
-    SyncIndex *pIndex = (SyncIndex *)key;
+    size_t    len;
+    void     *key = taosHashGetKey(pStub, &len);
+    uint64_t *pSeqNum = (uint64_t *)key;
 
     int64_t nowMS = taosGetTimestampMs();
     if (nowMS - pStub->createTime > ttl) {
-      taosArrayPush(delIndexArray, pIndex);
+      taosArrayPush(delIndexArray, pSeqNum);
       cnt++;
 
-      SSyncRaftEntry *pEntry = NULL;
-      int32_t         code = 0;
-      if (pSyncNode->pLogStore != NULL) {
-        code = pSyncNode->pLogStore->syncLogGetEntry(pSyncNode->pLogStore, *pIndex, &pEntry);
-        if (code == 0 && pEntry != NULL) {
-          SFsmCbMeta cbMeta = {0};
-          cbMeta.index = pEntry->index;
-          cbMeta.lastConfigIndex = syncNodeGetSnapshotConfigIndex(pSyncNode, cbMeta.index);
-          cbMeta.isWeak = pEntry->isWeak;
-          cbMeta.code = TSDB_CODE_SYN_TIMEOUT;
-          cbMeta.state = pSyncNode->state;
-          cbMeta.seqNum = pEntry->seqNum;
-          cbMeta.term = pEntry->term;
-          cbMeta.currentTerm = pSyncNode->pRaftStore->currentTerm;
-          cbMeta.flag = 0;
+      SFsmCbMeta cbMeta = {0};
+      cbMeta.index = SYNC_INDEX_INVALID;
+      cbMeta.lastConfigIndex = SYNC_INDEX_INVALID;
+      cbMeta.isWeak = false;
+      cbMeta.code = TSDB_CODE_SYN_TIMEOUT;
+      cbMeta.state = pSyncNode->state;
+      cbMeta.seqNum = *pSeqNum;
+      cbMeta.term = SYNC_TERM_INVALID;
+      cbMeta.currentTerm = pSyncNode->pRaftStore->currentTerm;
+      cbMeta.flag = 0;
 
-          SRpcMsg rpcMsg = pStub->rpcMsg;
-          rpcMsg.pCont = rpcMallocCont(pEntry->dataLen);
-          memcpy(rpcMsg.pCont, pEntry->data, pEntry->dataLen);
-          pSyncNode->pFsm->FpCommitCb(pSyncNode->pFsm, &rpcMsg, cbMeta);
-
-          syncEntryDestory(pEntry);
-        }
-      }
+      pSyncNode->pFsm->FpCommitCb(pSyncNode->pFsm, &(pStub->rpcMsg), cbMeta);
     }
 
     pStub = (SRespStub *)taosHashIterate(pObj->pRespHash, pStub);
   }
 
   int32_t arraySize = taosArrayGetSize(delIndexArray);
-  sDebug("vgId:%d, resp clean by ttl, cnt:%d, array-size:%d", pSyncNode->vgId, cnt, arraySize);
+  sDebug("vgId:%d, resp mgr clean by ttl, cnt:%d, array-size:%d", pSyncNode->vgId, cnt, arraySize);
 
   for (int32_t i = 0; i < arraySize; ++i) {
-    SyncIndex *pIndex = taosArrayGet(delIndexArray, i);
-    taosHashRemove(pObj->pRespHash, pIndex, sizeof(SyncIndex));
+    uint64_t *pSeqNum = taosArrayGet(delIndexArray, i);
+    taosHashRemove(pObj->pRespHash, pSeqNum, sizeof(uint64_t));
+    sDebug("vgId:%d, resp mgr clean by ttl, seq:%d", pSyncNode->vgId, *pSeqNum);
   }
   taosArrayDestroy(delIndexArray);
 }

source/libs/sync/src/syncTimeout.c

@@ -16,9 +16,11 @@
 #include "syncTimeout.h"
 #include "syncElection.h"
 #include "syncReplication.h"
+#include "syncRespMgr.h"
 
 int32_t syncNodeTimerRoutine(SSyncNode* ths) {
   syncNodeEventLog(ths, "timer routines ... ");
+  syncRespClean(ths->pSyncRespMgr);
   return 0;
 }