Merge pull request #18331 from taosdata/feature/sync-del-wal

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,26 +312,31 @@ int32_t syncBeginSnapshot(int64_t rid, int64_t lastApplyIndex) {
 _DEL_WAL:
 
   do {
-    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;
+    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();
+
+        code = walBeginSnapshot(pData->pWal, lastApplyIndex);
+        if (code == 0) {
+          sNTrace(pSyncNode, "wal snapshot begin, index:%" PRId64 ", last apply index:%" PRId64,
+                  pSyncNode->snapshottingIndex, lastApplyIndex);
+        } else {
+          sNError(pSyncNode, "wal snapshot begin error since:%s, index:%" PRId64 ", last apply index:%" PRId64,
+                  terrstr(terrno), pSyncNode->snapshottingIndex, lastApplyIndex);
+          atomic_store_64(&pSyncNode->snapshottingIndex, SYNC_INDEX_INVALID);
+        }
 
-      SSyncLogStoreData* pData = pSyncNode->pLogStore->data;
-      code = walBeginSnapshot(pData->pWal, lastApplyIndex);
-      if (code == 0) {
-        sNTrace(pSyncNode, "wal snapshot begin, index:%" PRId64 ", last apply index:%" PRId64,
-                pSyncNode->snapshottingIndex, lastApplyIndex);
       } else {
-        sNError(pSyncNode, "wal snapshot begin error since:%s, index:%" PRId64 ", last apply index:%" PRId64,
-                terrstr(terrno), pSyncNode->snapshottingIndex, lastApplyIndex);
-        atomic_store_64(&pSyncNode->snapshottingIndex, SYNC_INDEX_INVALID);
+        sNTrace(pSyncNode, "snapshotting for %" PRId64 ", do not delete wal for new-snapshot-index:%" PRId64,
+                snapshottingIndex, lastApplyIndex);
       }
-
-    } else {
-      sNTrace(pSyncNode, "snapshotting for %" PRId64 ", do not delete wal for new-snapshot-index:%" PRId64,
-              snapshottingIndex, lastApplyIndex);
     }
   } while (0);
 

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,18 +62,20 @@ static int32_t syncNodeTimerRoutine(SSyncNode* ths) {
     syncNodeCleanConfigIndex(ths);
   }
 
-  // end timeout wal snapshot
   int64_t timeNow = taosGetTimestampMs();
-  if (timeNow - ths->snapshottingIndex > SYNC_DEL_WAL_MS &&
-      atomic_load_64(&ths->snapshottingIndex) != SYNC_INDEX_INVALID) {
-    SSyncLogStoreData* pData = ths->pLogStore->data;
-    int32_t            code = walEndSnapshot(pData->pWal);
-    if (code != 0) {
-      sNError(ths, "timer wal snapshot end error since:%s", terrstr());
-      return -1;
-    } else {
-      sNTrace(ths, "wal snapshot end, index:%" PRId64, atomic_load_64(&ths->snapshottingIndex));
-      atomic_store_64(&ths->snapshottingIndex, SYNC_INDEX_INVALID);
+  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);
+      if (code != 0) {
+        sNError(ths, "timer wal snapshot end error since:%s", terrstr());
+        return -1;
+      } else {
+        sNTrace(ths, "wal snapshot end, index:%" PRId64, atomic_load_64(&ths->snapshottingIndex));
+        atomic_store_64(&ths->snapshottingIndex, SYNC_INDEX_INVALID);
+      }
     }
   }
 

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);
   }

source/libs/wal/src/walWrite.c

@@ -325,7 +325,8 @@ int32_t walEndSnapshot(SWal *pWal) {
   SWalFileInfo *pInfo = taosArraySearch(pWal->fileInfoSet, &tmp, compareWalFileInfo, TD_LE);
   if (pInfo) {
     if (ver >= pInfo->lastVer) {
-      pInfo--;
+      //pInfo--;
+      pInfo++;
     }
     if (POINTER_DISTANCE(pInfo, pWal->fileInfoSet->pData) > 0) {
       wDebug("vgId:%d, wal end remove for %" PRId64, pWal->cfg.vgId, pInfo->firstVer);