Merge pull request #14145 from taosdata/feature/3.0_mhli

fix(sync): update nextindex, matchindex when backto nolog

include/libs/sync/sync.h

@@ -26,8 +26,9 @@ extern "C" {
 
 extern bool gRaftDetailLog;
 
-#define SYNC_INDEX_BEGIN 0
+#define SYNC_INDEX_BEGIN   0
 #define SYNC_INDEX_INVALID -1
+#define SYNC_TERM_INVALID  0xFFFFFFFFFFFFFFFF
 
 typedef uint64_t SyncNodeId;
 typedef int32_t  SyncGroupId;
@@ -199,7 +200,7 @@ const char* syncGetMyRoleStr(int64_t rid);
 SyncTerm    syncGetMyTerm(int64_t rid);
 SyncGroupId syncGetVgId(int64_t rid);
 void        syncGetEpSet(int64_t rid, SEpSet* pEpSet);
-int32_t     syncPropose(int64_t rid, const SRpcMsg* pMsg, bool isWeak);
+int32_t     syncPropose(int64_t rid, SRpcMsg* pMsg, bool isWeak);
 bool        syncEnvIsStart();
 const char* syncStr(ESyncState state);
 bool        syncIsRestoreFinish(int64_t rid);

include/libs/sync/syncTools.h

@@ -43,7 +43,7 @@ void setElectTimerMS(int64_t rid, int32_t electTimerMS);
 void setHeartbeatTimerMS(int64_t rid, int32_t hbTimerMS);
 
 // for compatibility, the same as syncPropose
-int32_t syncForwardToPeer(int64_t rid, const SRpcMsg* pMsg, bool isWeak);
+int32_t syncForwardToPeer(int64_t rid, SRpcMsg* pMsg, bool isWeak);
 
 // utils
 const char* syncUtilState2String(ESyncState state);
@@ -468,7 +468,7 @@ typedef struct SyncLeaderTransfer {
    SRaftId  destId;
    */
   SNodeInfo newNodeInfo;
-  SRaftId newLeaderId;
+  SRaftId   newLeaderId;
 } SyncLeaderTransfer;
 
 SyncLeaderTransfer* syncLeaderTransferBuild(int32_t vgId);
@@ -489,17 +489,16 @@ void syncLeaderTransferPrint2(char* s, const SyncLeaderTransfer* pMsg);
 void syncLeaderTransferLog(const SyncLeaderTransfer* pMsg);
 void syncLeaderTransferLog2(char* s, const SyncLeaderTransfer* pMsg);
 
-
 // ---------------------------------------------
 typedef struct SyncReconfigFinish {
-  uint32_t bytes;
-  int32_t  vgId;
-  uint32_t msgType;
-  SSyncCfg oldCfg;
-  SSyncCfg newCfg;
+  uint32_t  bytes;
+  int32_t   vgId;
+  uint32_t  msgType;
+  SSyncCfg  oldCfg;
+  SSyncCfg  newCfg;
   SyncIndex newCfgIndex;
-  SyncTerm newCfgTerm;
-  uint64_t newCfgSeqNum;
+  SyncTerm  newCfgTerm;
+  uint64_t  newCfgSeqNum;
 
 } SyncReconfigFinish;
 
@@ -521,8 +520,6 @@ void syncReconfigFinishPrint2(char* s, const SyncReconfigFinish* pMsg);
 void syncReconfigFinishLog(const SyncReconfigFinish* pMsg);
 void syncReconfigFinishLog2(char* s, const SyncReconfigFinish* pMsg);
 
-
-
 // on message ----------------------
 int32_t syncNodeOnPingCb(SSyncNode* ths, SyncPing* pMsg);
 int32_t syncNodeOnPingReplyCb(SSyncNode* ths, SyncPingReply* pMsg);

source/libs/sync/src/syncAppendEntries.c

@@ -436,6 +436,11 @@ static bool syncNodeOnAppendEntriesLogOK(SSyncNode* pSyncNode, SyncAppendEntries
   }
 
   SyncTerm myPreLogTerm = syncNodeGetPreTerm(pSyncNode, pMsg->prevLogIndex + 1);
+  if (myPreLogTerm == SYNC_TERM_INVALID) {
+    sError("vgId:%d sync get pre term error, preindex:%ld", pSyncNode->vgId, pMsg->prevLogIndex);
+    return false;
+  }
+
   if (pMsg->prevLogIndex <= myLastIndex && pMsg->prevLogTerm == myPreLogTerm) {
     if (gRaftDetailLog) {
       sTrace(

source/libs/sync/src/syncMain.c

@@ -345,7 +345,7 @@ bool syncCanLeaderTransfer(int64_t rid) {
   return matchOK;
 }
 
-int32_t syncForwardToPeer(int64_t rid, const SRpcMsg* pMsg, bool isWeak) {
+int32_t syncForwardToPeer(int64_t rid, SRpcMsg* pMsg, bool isWeak) {
   int32_t ret = syncPropose(rid, pMsg, isWeak);
   return ret;
 }
@@ -584,7 +584,7 @@ void setHeartbeatTimerMS(int64_t rid, int32_t hbTimerMS) {
   taosReleaseRef(tsNodeRefId, pSyncNode->rid);
 }
 
-int32_t syncPropose(int64_t rid, const SRpcMsg* pMsg, bool isWeak) {
+int32_t syncPropose(int64_t rid, SRpcMsg* pMsg, bool isWeak) {
   int32_t ret = 0;
 
   SSyncNode* pSyncNode = taosAcquireRef(tsNodeRefId, rid);
@@ -1309,40 +1309,44 @@ void syncNodeEventLog(const SSyncNode* pSyncNode, char* str) {
   SyncIndex logLastIndex = pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore);
   SyncIndex logBeginIndex = pSyncNode->pLogStore->syncLogBeginIndex(pSyncNode->pLogStore);
 
+  char* pCfgStr = syncCfg2SimpleStr(&(pSyncNode->pRaftCfg->cfg));
+
   if (userStrLen < 256) {
-    char logBuf[128 + 256];
+    char logBuf[256 + 256];
     if (pSyncNode != NULL && pSyncNode->pRaftCfg != NULL && pSyncNode->pRaftStore != NULL) {
       snprintf(logBuf, sizeof(logBuf),
                "vgId:%d, sync %s %s, term:%lu, commit:%ld, beginlog:%ld, lastlog:%ld, lastsnapshot:%ld, standby:%d, "
                "replica-num:%d, "
-               "lconfig:%ld, changing:%d",
+               "lconfig:%ld, changing:%d, %s",
                pSyncNode->vgId, syncUtilState2String(pSyncNode->state), str, pSyncNode->pRaftStore->currentTerm,
                pSyncNode->commitIndex, logBeginIndex, logLastIndex, snapshot.lastApplyIndex,
                pSyncNode->pRaftCfg->isStandBy, pSyncNode->replicaNum, pSyncNode->pRaftCfg->lastConfigIndex,
-               pSyncNode->changing);
+               pSyncNode->changing, pCfgStr);
     } else {
       snprintf(logBuf, sizeof(logBuf), "%s", str);
     }
     sDebug("%s", logBuf);
 
   } else {
-    int   len = 128 + userStrLen;
+    int   len = 256 + userStrLen;
     char* s = (char*)taosMemoryMalloc(len);
     if (pSyncNode != NULL && pSyncNode->pRaftCfg != NULL && pSyncNode->pRaftStore != NULL) {
       snprintf(s, len,
                "vgId:%d, sync %s %s, term:%lu, commit:%ld, beginlog:%ld, lastlog:%ld, lastsnapshot:%ld, standby:%d, "
                "replica-num:%d, "
-               "lconfig:%ld, changing:%d",
+               "lconfig:%ld, changing:%d, %s",
                pSyncNode->vgId, syncUtilState2String(pSyncNode->state), str, pSyncNode->pRaftStore->currentTerm,
                pSyncNode->commitIndex, logBeginIndex, logLastIndex, snapshot.lastApplyIndex,
                pSyncNode->pRaftCfg->isStandBy, pSyncNode->replicaNum, pSyncNode->pRaftCfg->lastConfigIndex,
-               pSyncNode->changing);
+               pSyncNode->changing, pCfgStr);
     } else {
       snprintf(s, len, "%s", str);
     }
     sDebug("%s", s);
     taosMemoryFree(s);
   }
+
+  taosMemoryFree(pCfgStr);
 }
 
 void syncNodeErrorLog(const SSyncNode* pSyncNode, char* str) {
@@ -1455,6 +1459,17 @@ void syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* pNewConfig, SyncInde
     isAdd = false;
   }
 
+  // log begin config change
+  do {
+    char  eventLog[256];
+    char* pOldCfgStr = syncCfg2SimpleStr(&oldConfig);
+    char* pNewCfgStr = syncCfg2SimpleStr(pNewConfig);
+    snprintf(eventLog, sizeof(eventLog), "begin do config change, from %s to %s", pOldCfgStr, pNewCfgStr);
+    syncNodeEventLog(pSyncNode, eventLog);
+    taosMemoryFree(pOldCfgStr);
+    taosMemoryFree(pNewCfgStr);
+  } while (0);
+
   if (IamInNew) {
     pSyncNode->pRaftCfg->isStandBy = 0;  // change isStandBy to normal
   }
@@ -1613,6 +1628,17 @@ void syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* pNewConfig, SyncInde
   }
 
 _END:
+
+  // log end config change
+  do {
+    char  eventLog[256];
+    char* pOldCfgStr = syncCfg2SimpleStr(&oldConfig);
+    char* pNewCfgStr = syncCfg2SimpleStr(pNewConfig);
+    snprintf(eventLog, sizeof(eventLog), "end do config change, from %s to %s", pOldCfgStr, pNewCfgStr);
+    syncNodeEventLog(pSyncNode, eventLog);
+    taosMemoryFree(pOldCfgStr);
+    taosMemoryFree(pNewCfgStr);
+  } while (0);
   return;
 }
 
@@ -1888,6 +1914,16 @@ SyncIndex syncNodeSyncStartIndex(SSyncNode* pSyncNode) {
   return syncStartIndex;
 }
 
+SyncIndex syncNodeGetPreIndex(SSyncNode* pSyncNode, SyncIndex index) {
+  SyncIndex preIndex = index - 1;
+  if (preIndex < SYNC_INDEX_INVALID) {
+    preIndex = SYNC_INDEX_INVALID;
+  }
+
+  return preIndex;
+}
+
+/*
 SyncIndex syncNodeGetPreIndex(SSyncNode* pSyncNode, SyncIndex index) {
   ASSERT(index >= SYNC_INDEX_BEGIN);
 
@@ -1900,7 +1936,42 @@ SyncIndex syncNodeGetPreIndex(SSyncNode* pSyncNode, SyncIndex index) {
   SyncIndex preIndex = index - 1;
   return preIndex;
 }
+*/
+
+SyncTerm syncNodeGetPreTerm(SSyncNode* pSyncNode, SyncIndex index) {
+  if (index < SYNC_INDEX_BEGIN) {
+    return SYNC_TERM_INVALID;
+  }
+
+  if (index == SYNC_INDEX_BEGIN) {
+    return 0;
+  }
+
+  SyncTerm        preTerm = 0;
+  SyncIndex       preIndex = index - 1;
+  SSyncRaftEntry* pPreEntry = NULL;
+  int32_t         code = pSyncNode->pLogStore->syncLogGetEntry(pSyncNode->pLogStore, preIndex, &pPreEntry);
+  if (code == 0) {
+    ASSERT(pPreEntry != NULL);
+    preTerm = pPreEntry->term;
+    taosMemoryFree(pPreEntry);
+    return preTerm;
+  } else {
+    if (terrno == TSDB_CODE_WAL_LOG_NOT_EXIST) {
+      SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0, .lastConfigIndex = -1};
+      if (pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
+        pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
+        if (snapshot.lastApplyIndex == preIndex) {
+          return snapshot.lastApplyTerm;
+        }
+      }
+    }
+  }
+
+  return SYNC_TERM_INVALID;
+}
 
+#if 0
 SyncTerm syncNodeGetPreTerm(SSyncNode* pSyncNode, SyncIndex index) {
   ASSERT(index >= SYNC_INDEX_BEGIN);
 
@@ -1938,6 +2009,7 @@ SyncTerm syncNodeGetPreTerm(SSyncNode* pSyncNode, SyncIndex index) {
   ASSERT(0);
   return -1;
 }
+#endif
 
 #if 0
 SyncTerm syncNodeGetPreTerm(SSyncNode* pSyncNode, SyncIndex index) {

source/libs/sync/src/syncRaftLog.c

@@ -144,7 +144,8 @@ static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntr
 
   SyncIndex writeIndex = raftLogWriteIndex(pLogStore);
   if (pEntry->index != writeIndex) {
-    sError("raftLogAppendEntry error, pEntry->index:%ld update to writeIndex:%ld", pEntry->index, writeIndex);
+    sError("vgId:%d wal write index error, entry-index:%ld update to %ld", pData->pSyncNode->vgId, pEntry->index,
+           writeIndex);
     pEntry->index = writeIndex;
   }
 
@@ -157,10 +158,10 @@ static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntr
   if (code != 0) {
     int32_t     err = terrno;
     const char* errStr = tstrerror(err);
-    int32_t     linuxErr = errno;
-    const char* linuxErrMsg = strerror(errno);
-    sError("raftLogAppendEntry error, err:%d %X, msg:%s, linuxErr:%d, linuxErrMsg:%s", err, err, errStr, linuxErr,
-           linuxErrMsg);
+    int32_t     sysErr = errno;
+    const char* sysErrStr = strerror(errno);
+    sError("vgId:%d wal write error, index:%ld, err:%d %X, msg:%s, syserr:%d, sysmsg:%s", pData->pSyncNode->vgId,
+           pEntry->index, err, err, errStr, sysErr, sysErrStr);
     ASSERT(0);
   }
 
@@ -237,12 +238,15 @@ static int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index,
   if (code != 0) {
     int32_t     err = terrno;
     const char* errStr = tstrerror(err);
-    int32_t     linuxErr = errno;
-    const char* linuxErrMsg = strerror(errno);
-    sError("raftLogGetEntry error, err:%d %X, msg:%s, linuxErr:%d, linuxErrMsg:%s", err, err, errStr, linuxErr,
-           linuxErrMsg);
+    int32_t     sysErr = errno;
+    const char* sysErrStr = strerror(errno);
+    sError("vgId:%d wal read error, index:%ld, err:%d %X, msg:%s, syserr:%d, sysmsg:%s", pData->pSyncNode->vgId, index,
+           err, err, errStr, sysErr, sysErrStr);
 
+    int32_t saveErr = terrno;
     walCloseReadHandle(pWalHandle);
+    terrno = saveErr;
+
     return code;
   }
 
@@ -257,8 +261,9 @@ static int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index,
   ASSERT((*ppEntry)->dataLen == pWalHandle->pHead->head.bodyLen);
   memcpy((*ppEntry)->data, pWalHandle->pHead->head.body, pWalHandle->pHead->head.bodyLen);
 
-  // need to hold, do not new every time!!
+  int32_t saveErr = terrno;
   walCloseReadHandle(pWalHandle);
+  terrno = saveErr;
 
   return code;
 }
@@ -270,10 +275,11 @@ static int32_t raftLogTruncate(struct SSyncLogStore* pLogStore, SyncIndex fromIn
   if (code != 0) {
     int32_t     err = terrno;
     const char* errStr = tstrerror(err);
-    int32_t     linuxErr = errno;
-    const char* linuxErrMsg = strerror(errno);
-    sError("raftLogTruncate error, err:%d %X, msg:%s, linuxErr:%d, linuxErrMsg:%s", err, err, errStr, linuxErr,
-           linuxErrMsg);
+    int32_t     sysErr = errno;
+    const char* sysErrStr = strerror(errno);
+    sError("vgId:%d wal truncate error, from-index:%ld, err:%d %X, msg:%s, syserr:%d, sysmsg:%s",
+           pData->pSyncNode->vgId, fromIndex, err, err, errStr, sysErr, sysErrStr);
+
     ASSERT(0);
   }
   return code;
@@ -360,10 +366,11 @@ int32_t logStoreAppendEntry(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry) {
   if (code != 0) {
     int32_t     err = terrno;
     const char* errStr = tstrerror(err);
-    int32_t     linuxErr = errno;
-    const char* linuxErrMsg = strerror(errno);
-    sError("walWriteWithSyncInfo error, err:%d %X, msg:%s, linuxErr:%d, linuxErrMsg:%s", err, err, errStr, linuxErr,
-           linuxErrMsg);
+    int32_t     sysErr = errno;
+    const char* sysErrStr = strerror(errno);
+    sError("vgId:%d wal write error, index:%ld, err:%d %X, msg:%s, syserr:%d, sysmsg:%s", pData->pSyncNode->vgId,
+           pEntry->index, err, err, errStr, sysErr, sysErrStr);
+
     ASSERT(0);
   }
 
@@ -389,10 +396,11 @@ SSyncRaftEntry* logStoreGetEntry(SSyncLogStore* pLogStore, SyncIndex index) {
     if (code != 0) {
       int32_t     err = terrno;
       const char* errStr = tstrerror(err);
-      int32_t     linuxErr = errno;
-      const char* linuxErrMsg = strerror(errno);
-      sError("walReadWithHandle error, err:%d %X, msg:%s, linuxErr:%d, linuxErrMsg:%s", err, err, errStr, linuxErr,
-             linuxErrMsg);
+      int32_t     sysErr = errno;
+      const char* sysErrStr = strerror(errno);
+      sError("vgId:%d wal read error, index:%ld, err:%d %X, msg:%s, syserr:%d, sysmsg:%s", pData->pSyncNode->vgId,
+             index, err, err, errStr, sysErr, sysErrStr);
+
       ASSERT(0);
     }
     // ASSERT(walReadWithHandle(pWalHandle, index) == 0);
@@ -409,8 +417,10 @@ SSyncRaftEntry* logStoreGetEntry(SSyncLogStore* pLogStore, SyncIndex index) {
     ASSERT(pEntry->dataLen == pWalHandle->pHead->head.bodyLen);
     memcpy(pEntry->data, pWalHandle->pHead->head.body, pWalHandle->pHead->head.bodyLen);
 
-    // need to hold, do not new every time!!
+    int32_t saveErr = terrno;
     walCloseReadHandle(pWalHandle);
+    terrno = saveErr;
+
     return pEntry;
 
   } else {
@@ -426,10 +436,11 @@ int32_t logStoreTruncate(SSyncLogStore* pLogStore, SyncIndex fromIndex) {
   if (code != 0) {
     int32_t     err = terrno;
     const char* errStr = tstrerror(err);
-    int32_t     linuxErr = errno;
-    const char* linuxErrMsg = strerror(errno);
-    sError("walRollback error, err:%d %X, msg:%s, linuxErr:%d, linuxErrMsg:%s", err, err, errStr, linuxErr,
-           linuxErrMsg);
+    int32_t     sysErr = errno;
+    const char* sysErrStr = strerror(errno);
+    sError("vgId:%d wal truncate error, from-index:%ld, err:%d %X, msg:%s, syserr:%d, sysmsg:%s",
+           pData->pSyncNode->vgId, fromIndex, err, err, errStr, sysErr, sysErrStr);
+
     ASSERT(0);
   }
   return 0;
@@ -460,9 +471,11 @@ int32_t logStoreUpdateCommitIndex(SSyncLogStore* pLogStore, SyncIndex index) {
   if (code != 0) {
     int32_t     err = terrno;
     const char* errStr = tstrerror(err);
-    int32_t     linuxErr = errno;
-    const char* linuxErrMsg = strerror(errno);
-    sError("walCommit error, err:%d %X, msg:%s, linuxErr:%d, linuxErrMsg:%s", err, err, errStr, linuxErr, linuxErrMsg);
+    int32_t     sysErr = errno;
+    const char* sysErrStr = strerror(errno);
+    sError("vgId:%d wal update commit index error, index:%ld, err:%d %X, msg:%s, syserr:%d, sysmsg:%s",
+           pData->pSyncNode->vgId, index, err, err, errStr, sysErr, sysErrStr);
+
     ASSERT(0);
   }
   return 0;

source/libs/sync/src/syncReplication.c

@@ -139,6 +139,15 @@ int32_t syncNodeAppendEntriesPeersSnapshot(SSyncNode* pSyncNode) {
     // pre index, pre term
     SyncIndex preLogIndex = syncNodeGetPreIndex(pSyncNode, nextIndex);
     SyncTerm  preLogTerm = syncNodeGetPreTerm(pSyncNode, nextIndex);
+    if (preLogTerm == SYNC_TERM_INVALID) {
+      SyncIndex newNextIndex = syncNodeGetLastIndex(pSyncNode) + 1;
+      syncIndexMgrSetIndex(pSyncNode->pNextIndex, pDestId, newNextIndex);
+      syncIndexMgrSetIndex(pSyncNode->pMatchIndex, pDestId, SYNC_INDEX_INVALID);
+      sError("vgId:%d sync get pre term error, nextIndex:%ld, update next-index:%ld, match-index:%d, raftid:%ld",
+             pSyncNode->vgId, nextIndex, newNextIndex, SYNC_INDEX_INVALID, pDestId->addr);
+
+      return -1;
+    }
 
     // batch optimized
     // SyncIndex lastIndex = syncUtilMinIndex(pSyncNode->pLogStore->getLastIndex(pSyncNode->pLogStore), nextIndex);

source/libs/sync/src/syncSnapshot.c

@@ -113,8 +113,16 @@ void snapshotSenderStart(SSyncSnapshotSender *pSender, SSnapshot snapshot, void
     }
 
     if (!getLastConfig) {
-      syncNodeLog3("", pSender->pSyncNode);
-      ASSERT(0);
+      char logBuf[128];
+      snprintf(logBuf, sizeof(logBuf), "snapshot sender update lcindex from %ld to -1",
+               pSender->snapshot.lastConfigIndex);
+      pSender->snapshot.lastConfigIndex = -1;
+
+      char *eventLog = snapshotSender2SimpleStr(pSender, logBuf);
+      syncNodeEventLog(pSender->pSyncNode, eventLog);
+      taosMemoryFree(eventLog);
+
+      memset(&(pSender->lastConfig), 0, sizeof(SSyncCfg));
     }
 
   } else {