refactor(sync): del wal in multi-replicas

include/libs/sync/sync.h

@@ -36,7 +36,7 @@ extern "C" {
 #define SYNC_DEL_WAL_MS              (1000 * 60)
 #define SYNC_ADD_QUORUM_COUNT        3
 #define SYNC_MNODE_LOG_RETENTION     10000
-#define SYNC_VNODE_LOG_RETENTION     100
+#define SYNC_VNODE_LOG_RETENTION     20
 #define SNAPSHOT_MAX_CLOCK_SKEW_MS   1000 * 10
 #define SNAPSHOT_WAIT_MS             1000 * 30
 

source/libs/sync/src/syncAppendEntriesReply.c

@@ -67,6 +67,9 @@ int32_t syncNodeOnAppendEntriesReply(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
       if (pMsg->matchIndex > oldMatchIndex) {
         syncIndexMgrSetIndex(ths->pMatchIndex, &(pMsg->srcId), pMsg->matchIndex);
         syncMaybeAdvanceCommitIndex(ths);
+
+        // maybe update minMatchIndex
+        ths->minMatchIndex = syncMinMatchIndex(ths);
       }
       syncIndexMgrSetIndex(ths->pNextIndex, &(pMsg->srcId), pMsg->matchIndex + 1);
 

source/libs/sync/src/syncMain.c

@@ -243,6 +243,18 @@ int32_t syncBeginSnapshot(int64_t rid, int64_t lastApplyIndex) {
     goto _DEL_WAL;
 
   } else {
+    lastApplyIndex -= SYNC_VNODE_LOG_RETENTION;
+
+    SyncIndex beginIndex = pSyncNode->pLogStore->syncLogBeginIndex(pSyncNode->pLogStore);
+    SyncIndex endIndex = pSyncNode->pLogStore->syncLogEndIndex(pSyncNode->pLogStore);
+    bool      isEmpty = pSyncNode->pLogStore->syncLogIsEmpty(pSyncNode->pLogStore);
+
+    if (isEmpty || !(lastApplyIndex >= beginIndex && lastApplyIndex <= endIndex)) {
+      sNTrace(pSyncNode, "new-snapshot-index:%" PRId64 ", empty:%d, do not delete wal", lastApplyIndex, isEmpty);
+      syncNodeRelease(pSyncNode);
+      return 0;
+    }
+
     // vnode
     if (pSyncNode->replicaNum > 1) {
       // multi replicas
@@ -300,13 +312,17 @@ int32_t syncBeginSnapshot(int64_t rid, int64_t lastApplyIndex) {
 _DEL_WAL:
 
   do {
+    SSyncLogStoreData* pData = pSyncNode->pLogStore->data;
+    SyncIndex          snapshotVer = walGetSnapshotVer(pData->pWal);
+    SyncIndex          walCommitVer = walGetCommittedVer(pData->pWal);
+    SyncIndex          wallastVer = walGetLastVer(pData->pWal);
+    if (lastApplyIndex <= walCommitVer) {
       SyncIndex snapshottingIndex = atomic_load_64(&pSyncNode->snapshottingIndex);
 
       if (snapshottingIndex == SYNC_INDEX_INVALID) {
         atomic_store_64(&pSyncNode->snapshottingIndex, lastApplyIndex);
         pSyncNode->snapshottingTime = taosGetTimestampMs();
 
-      SSyncLogStoreData* pData = pSyncNode->pLogStore->data;
         code = walBeginSnapshot(pData->pWal, lastApplyIndex);
         if (code == 0) {
           sNTrace(pSyncNode, "wal snapshot begin, index:%" PRId64 ", last apply index:%" PRId64,
@@ -321,6 +337,7 @@ _DEL_WAL:
         sNTrace(pSyncNode, "snapshotting for %" PRId64 ", do not delete wal for new-snapshot-index:%" PRId64,
                 snapshottingIndex, lastApplyIndex);
       }
+    }
   } while (0);
 
   syncNodeRelease(pSyncNode);

source/libs/sync/src/syncRaftLog.c

@@ -375,7 +375,17 @@ static int32_t raftLogGetLastEntry(SSyncLogStore* pLogStore, SSyncRaftEntry** pp
 int32_t raftLogUpdateCommitIndex(SSyncLogStore* pLogStore, SyncIndex index) {
   SSyncLogStoreData* pData = pLogStore->data;
   SWal*              pWal = pData->pWal;
-  // ASSERT(walCommit(pWal, index) == 0);
+
+  // need not update
+  SyncIndex snapshotVer = walGetSnapshotVer(pWal);
+  SyncIndex walCommitVer = walGetCommittedVer(pWal);
+  SyncIndex wallastVer = walGetLastVer(pWal);
+
+  if (index < snapshotVer || index > wallastVer) {
+    // ignore
+    return 0;
+  }
+
   int32_t code = walCommit(pWal, index);
   if (code != 0) {
     int32_t     err = terrno;

source/libs/sync/src/syncTimeout.c

@@ -62,9 +62,10 @@ static int32_t syncNodeTimerRoutine(SSyncNode* ths) {
     syncNodeCleanConfigIndex(ths);
   }
 
-  // end timeout wal snapshot
   int64_t timeNow = taosGetTimestampMs();
-  if (timeNow - ths->snapshottingIndex > SYNC_DEL_WAL_MS &&
+  if (atomic_load_64(&ths->snapshottingIndex) != SYNC_INDEX_INVALID) {
+    // end timeout wal snapshot
+    if (timeNow - ths->snapshottingTime > SYNC_DEL_WAL_MS &&
         atomic_load_64(&ths->snapshottingIndex) != SYNC_INDEX_INVALID) {
       SSyncLogStoreData* pData = ths->pLogStore->data;
       int32_t            code = walEndSnapshot(pData->pWal);
@@ -76,6 +77,7 @@ static int32_t syncNodeTimerRoutine(SSyncNode* ths) {
         atomic_store_64(&ths->snapshottingIndex, SYNC_INDEX_INVALID);
       }
     }
+  }
 
 #if 0
   if (!syncNodeIsMnode(ths)) {

source/libs/sync/src/syncUtil.c

@@ -239,11 +239,11 @@ void syncPrintNodeLog(const char* flags, ELogLevel level, int32_t dflag, SSyncNo
                  "vgId:%d, sync %s "
                  "%s"
                  ", tm:%" PRIu64 ", cmt:%" PRId64 ", fst:%" PRId64 ", lst:%" PRId64 ", min:%" PRId64 ", snap:%" PRId64
-                 ", snap-tm:%" PRIu64 ", sby:%d, aq:%d, bch:%d, r-num:%d, lcfg:%" PRId64
+                 ", snap-tm:%" PRIu64 ", sby:%d, aq:%d, snaping:%" PRId64 ", r-num:%d, lcfg:%" PRId64
                  ", chging:%d, rsto:%d, dquorum:%d, elt:%" PRId64 ", hb:%" PRId64 ", %s, %s",
                  pNode->vgId, syncStr(pNode->state), eventLog, currentTerm, pNode->commitIndex, logBeginIndex,
                  logLastIndex, pNode->minMatchIndex, snapshot.lastApplyIndex, snapshot.lastApplyTerm,
-                 pNode->pRaftCfg->isStandBy, aqItems, pNode->pRaftCfg->batchSize, pNode->replicaNum,
+                 pNode->pRaftCfg->isStandBy, aqItems, pNode->snapshottingIndex, pNode->replicaNum,
                  pNode->pRaftCfg->lastConfigIndex, pNode->changing, pNode->restoreFinish, quorum,
                  pNode->electTimerLogicClock, pNode->heartbeatTimerLogicClockUser, peerStr, cfgStr);
   }