From bd492f790efc0c8278c7bebb0f21c606d823bb72 Mon Sep 17 00:00:00 2001
From: dmchen <cademfly@hotmail.com>
Date: Wed, 26 Feb 2025 16:55:06 +0800
Subject: [PATCH] fix/remove-sync-heartbeat-lock

---
 source/libs/sync/inc/syncPipeline.h |  1 +
 source/libs/sync/src/syncPipeline.c | 16 +++++++++++++---
 2 files changed, 14 insertions(+), 3 deletions(-)

diff --git a/source/libs/sync/inc/syncPipeline.h b/source/libs/sync/inc/syncPipeline.h
index eeb24d2f16..147f8a67ae 100644
--- a/source/libs/sync/inc/syncPipeline.h
+++ b/source/libs/sync/inc/syncPipeline.h
@@ -40,6 +40,7 @@ typedef struct SSyncLogReplMgr {
   int32_t       retryBackoff;
   int32_t       peerId;
   int32_t       sendCount;
+  TdThreadMutex mutex;
 } SSyncLogReplMgr;
 
 typedef struct SSyncLogBufEntry {
diff --git a/source/libs/sync/src/syncPipeline.c b/source/libs/sync/src/syncPipeline.c
index 18252db9ee..ceaba1e643 100644
--- a/source/libs/sync/src/syncPipeline.c
+++ b/source/libs/sync/src/syncPipeline.c
@@ -1140,26 +1140,29 @@ int32_t syncLogReplRecover(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncAppendEn
 
 int32_t syncLogReplProcessHeartbeatReply(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncHeartbeatReply* pMsg) {
   SSyncLogBuffer* pBuf = pNode->pLogBuf;
-  (void)taosThreadMutexLock(&pBuf->mutex);
+  (void)taosThreadMutexLock(&pMgr->mutex);
   if (pMsg->startTime != 0 && pMsg->startTime != pMgr->peerStartTime) {
     sInfo("vgId:%d, reset sync log repl in heartbeat. peer:%" PRIx64 ", start time:%" PRId64 ", old:%" PRId64 "",
           pNode->vgId, pMsg->srcId.addr, pMsg->startTime, pMgr->peerStartTime);
     syncLogReplReset(pMgr);
     pMgr->peerStartTime = pMsg->startTime;
   }
-  (void)taosThreadMutexUnlock(&pBuf->mutex);
+  (void)taosThreadMutexUnlock(&pMgr->mutex);
   return 0;
 }
 
 int32_t syncLogReplProcessReply(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncAppendEntriesReply* pMsg) {
   SSyncLogBuffer* pBuf = pNode->pLogBuf;
-  (void)taosThreadMutexLock(&pBuf->mutex);
+  (void)taosThreadMutexLock(&pMgr->mutex);
   if (pMsg->startTime != pMgr->peerStartTime) {
     sInfo("vgId:%d, reset sync log repl in appendlog reply. peer:%" PRIx64 ", start time:%" PRId64 ", old:%" PRId64,
           pNode->vgId, pMsg->srcId.addr, pMsg->startTime, pMgr->peerStartTime);
     syncLogReplReset(pMgr);
     pMgr->peerStartTime = pMsg->startTime;
   }
+  (void)taosThreadMutexUnlock(&pMgr->mutex);
+
+  (void)taosThreadMutexLock(&pBuf->mutex);
 
   int32_t code = 0;
   if (pMgr->restored) {
@@ -1324,6 +1327,12 @@ SSyncLogReplMgr* syncLogReplCreate() {
     return NULL;
   }
 
+  int32_t code = taosThreadMutexInit(&pMgr->mutex, NULL);
+  if (code) {
+    terrno = code;
+    return NULL;
+  }
+
   return pMgr;
 }
 
@@ -1331,6 +1340,7 @@ void syncLogReplDestroy(SSyncLogReplMgr* pMgr) {
   if (pMgr == NULL) {
     return;
   }
+  taosThreadMutexDestroy(&pMgr->mutex);
   taosMemoryFree(pMgr);
   return;
 }