refactor(sync): persist batch size

source/libs/sync/inc/syncRaftCfg.h

@@ -36,6 +36,7 @@ typedef struct SRaftCfg {
   TdFilePtr pFile;
   char      path[TSDB_FILENAME_LEN * 2];
   int8_t    isStandBy;
+  int32_t   batchSize;
   int8_t    snapshotStrategy;
   SyncIndex lastConfigIndex;
 
@@ -62,6 +63,7 @@ int32_t raftCfgFromStr(const char *s, SRaftCfg *pRaftCfg);
 
 typedef struct SRaftCfgMeta {
   int8_t    isStandBy;
+  int32_t   batchSize;
   int8_t    snapshotStrategy;
   SyncIndex lastConfigIndex;
 } SRaftCfgMeta;

source/libs/sync/src/syncAppendEntries.c

@@ -834,7 +834,7 @@ int32_t syncNodeOnAppendEntriesSnapshot2Cb(SSyncNode* ths, SyncAppendEntriesBatc
   //
   // operation:
   // if hasAppendEntries && pMsg->prevLogIndex == ths->commitIndex, append entry
-  // match my-commit-index or my-commit-index + 1
+  // match my-commit-index or my-commit-index + batchSize
   do {
     bool condition = (pMsg->term == ths->pRaftStore->currentTerm) && (ths->state == TAOS_SYNC_STATE_FOLLOWER) &&
                      (pMsg->prevLogIndex <= ths->commitIndex);
@@ -928,11 +928,13 @@ int32_t syncNodeOnAppendEntriesSnapshot2Cb(SSyncNode* ths, SyncAppendEntriesBatc
     bool condition = condition1 || condition2;
 
     if (condition) {
-      char logBuf[128];
-      snprintf(logBuf, sizeof(logBuf),
-               "recv sync-append-entries-batch, not match, pre-index:%ld, pre-term:%lu, datalen:%d", pMsg->prevLogIndex,
-               pMsg->prevLogTerm, pMsg->dataLen);
-      syncNodeEventLog(ths, logBuf);
+      do {
+        char logBuf[128];
+        snprintf(logBuf, sizeof(logBuf),
+                 "recv sync-append-entries-batch, not match, pre-index:%ld, pre-term:%lu, datalen:%d",
+                 pMsg->prevLogIndex, pMsg->prevLogTerm, pMsg->dataLen);
+        syncNodeEventLog(ths, logBuf);
+      } while (0);
 
       // prepare response msg
       SyncAppendEntriesReply* pReply = syncAppendEntriesReplyBuild(ths->vgId);

source/libs/sync/src/syncAppendEntriesReply.c

@@ -109,19 +109,30 @@ int32_t syncNodeOnAppendEntriesReplyCb(SSyncNode* ths, SyncAppendEntriesReply* p
 }
 
 // only start once
-static void syncNodeStartSnapshot(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex, SyncTerm lastApplyTerm,
-                                  SyncAppendEntriesReply* pMsg) {
+static void syncNodeStartSnapshotOnce(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex, SyncTerm lastApplyTerm,
+                                      SyncAppendEntriesReply* pMsg) {
   // get sender
   SSyncSnapshotSender* pSender = syncNodeGetSnapshotSender(ths, &(pMsg->srcId));
   ASSERT(pSender != NULL);
 
+  if (snapshotSenderIsStart(pSender)) {
+    do {
+      char* eventLog = snapshotSender2SimpleStr(pSender, "snapshot sender already start");
+      syncNodeErrorLog(ths, eventLog);
+      taosMemoryFree(eventLog);
+    } while (0);
+
+    return;
+  }
+
   SSnapshot snapshot = {
       .data = NULL, .lastApplyIndex = endIndex, .lastApplyTerm = lastApplyTerm, .lastConfigIndex = SYNC_INDEX_INVALID};
-
   void*          pReader = NULL;
   SSnapshotParam readerParam = {.start = beginIndex, .end = endIndex};
-  ths->pFsm->FpSnapshotStartRead(ths->pFsm, &readerParam, &pReader);
-  if (!snapshotSenderIsStart(pSender) && pMsg->privateTerm < pSender->privateTerm) {
+  int32_t        code = ths->pFsm->FpSnapshotStartRead(ths->pFsm, &readerParam, &pReader);
+  ASSERT(code == 0);
+
+  if (pMsg->privateTerm < pSender->privateTerm) {
     ASSERT(pReader != NULL);
     snapshotSenderStart(pSender, readerParam, snapshot, pReader);
 
@@ -178,7 +189,9 @@ int32_t syncNodeOnAppendEntriesReplySnapshot2Cb(SSyncNode* ths, SyncAppendEntrie
       // start snapshot <match+1, old snapshot.end>
       SSnapshot oldSnapshot;
       ths->pFsm->FpGetSnapshotInfo(ths->pFsm, &oldSnapshot);
-      syncNodeStartSnapshot(ths, newMatchIndex + 1, oldSnapshot.lastApplyIndex, oldSnapshot.lastApplyTerm, pMsg);
+      ASSERT(oldSnapshot.lastApplyIndex >= newMatchIndex + 1);
+      syncNodeStartSnapshotOnce(ths, newMatchIndex + 1, oldSnapshot.lastApplyIndex, oldSnapshot.lastApplyTerm,
+                                pMsg);  // term maybe not ok?
 
       syncIndexMgrSetIndex(ths->pNextIndex, &(pMsg->srcId), oldSnapshot.lastApplyIndex + 1);
       syncIndexMgrSetIndex(ths->pMatchIndex, &(pMsg->srcId), newMatchIndex);
@@ -187,7 +200,6 @@ int32_t syncNodeOnAppendEntriesReplySnapshot2Cb(SSyncNode* ths, SyncAppendEntrie
   } else {
     SyncIndex nextIndex = syncIndexMgrGetIndex(ths->pNextIndex, &(pMsg->srcId));
 
-    // notice! int64, uint64
     if (nextIndex > SYNC_INDEX_BEGIN) {
       --nextIndex;
 
@@ -198,7 +210,7 @@ int32_t syncNodeOnAppendEntriesReplySnapshot2Cb(SSyncNode* ths, SyncAppendEntrie
         SSyncRaftEntry* pEntry;
         int32_t         code = ths->pLogStore->syncLogGetEntry(ths->pLogStore, nextIndex, &pEntry);
         ASSERT(code == 0);
-        syncNodeStartSnapshot(ths, SYNC_INDEX_BEGIN, nextIndex, pEntry->term, pMsg);
+        syncNodeStartSnapshotOnce(ths, SYNC_INDEX_BEGIN, nextIndex, pEntry->term, pMsg);
 
         // get sender
         SSyncSnapshotSender* pSender = syncNodeGetSnapshotSender(ths, &(pMsg->srcId));

source/libs/sync/src/syncMain.c

@@ -846,6 +846,7 @@ SSyncNode* syncNodeOpen(const SSyncInfo* pOldSyncInfo) {
     meta.isStandBy = pSyncInfo->isStandBy;
     meta.snapshotStrategy = pSyncInfo->snapshotStrategy;
     meta.lastConfigIndex = SYNC_INDEX_INVALID;
+    meta.batchSize = pSyncInfo->batchSize;
     ret = raftCfgCreateFile((SSyncCfg*)&(pSyncInfo->syncCfg), meta, pSyncNode->configPath);
     ASSERT(ret == 0);
 

source/libs/sync/src/syncRaftCfg.c

@@ -183,6 +183,7 @@ cJSON *raftCfg2Json(SRaftCfg *pRaftCfg) {
   cJSON_AddItemToObject(pRoot, "SSyncCfg", syncCfg2Json(&(pRaftCfg->cfg)));
   cJSON_AddNumberToObject(pRoot, "isStandBy", pRaftCfg->isStandBy);
   cJSON_AddNumberToObject(pRoot, "snapshotStrategy", pRaftCfg->snapshotStrategy);
+  cJSON_AddNumberToObject(pRoot, "batchSize", pRaftCfg->batchSize);
 
   char buf64[128];
   snprintf(buf64, sizeof(buf64), "%ld", pRaftCfg->lastConfigIndex);
@@ -228,6 +229,7 @@ int32_t raftCfgCreateFile(SSyncCfg *pCfg, SRaftCfgMeta meta, const char *path) {
   SRaftCfg raftCfg;
   raftCfg.cfg = *pCfg;
   raftCfg.isStandBy = meta.isStandBy;
+  raftCfg.batchSize = meta.batchSize;
   raftCfg.snapshotStrategy = meta.snapshotStrategy;
   raftCfg.lastConfigIndex = meta.lastConfigIndex;
   raftCfg.configIndexCount = 1;
@@ -257,6 +259,9 @@ int32_t raftCfgFromJson(const cJSON *pRoot, SRaftCfg *pRaftCfg) {
   cJSON *pJsonIsStandBy = cJSON_GetObjectItem(pJson, "isStandBy");
   pRaftCfg->isStandBy = cJSON_GetNumberValue(pJsonIsStandBy);
 
+  cJSON *pJsonBatchSize = cJSON_GetObjectItem(pJson, "batchSize");
+  pRaftCfg->batchSize = cJSON_GetNumberValue(pJsonBatchSize);
+
   cJSON *pJsonSnapshotStrategy = cJSON_GetObjectItem(pJson, "snapshotStrategy");
   pRaftCfg->snapshotStrategy = cJSON_GetNumberValue(pJsonSnapshotStrategy);
 

source/libs/sync/test/syncRaftCfgTest.cpp

@@ -26,6 +26,7 @@ SRaftCfg* createRaftCfg() {
     snprintf(((pCfg->cfg.nodeInfo)[i]).nodeFqdn, sizeof(((pCfg->cfg.nodeInfo)[i]).nodeFqdn), "100.200.300.%d", i);
   }
   pCfg->isStandBy = taosGetTimestampSec() % 100;
+  pCfg->batchSize = taosGetTimestampSec() % 100;
 
   pCfg->configIndexCount = 5;
   for (int i = 0; i < MAX_CONFIG_INDEX_COUNT; ++i) {
@@ -84,6 +85,7 @@ void test3() {
     SRaftCfgMeta meta;
     meta.isStandBy = 7;
     meta.snapshotStrategy = 9;
+    meta.batchSize = 10;
     meta.lastConfigIndex = 789;
     raftCfgCreateFile(pCfg, meta, s);
     printf("%s create json file: %s \n", (char*)__FUNCTION__, s);
@@ -109,6 +111,7 @@ void test5() {
   pCfg->cfg.myIndex = taosGetTimestampSec();
   pCfg->isStandBy += 2;
   pCfg->snapshotStrategy += 3;
+  pCfg->batchSize += 4;
   pCfg->lastConfigIndex += 1000;
 
   pCfg->configIndexCount = 5;