From b6aea6ae17eb8a718a8a597143268f2d63124935 Mon Sep 17 00:00:00 2001
From: Minghao Li <castermode@gmail.com>
Date: Mon, 7 Nov 2022 16:40:53 +0800
Subject: [PATCH] fix(sync): when apply queue not empty, can not read

---
 include/libs/sync/sync.h               |  1 +
 source/dnode/mnode/impl/src/mndSync.c  |  8 ++++++++
 source/dnode/vnode/src/vnd/vnodeSync.c |  7 +++++++
 source/libs/sync/src/syncMain.c        | 24 +++++++++++++++---------
 4 files changed, 31 insertions(+), 9 deletions(-)

diff --git a/include/libs/sync/sync.h b/include/libs/sync/sync.h
index 8db3d89e39..74a73f6b10 100644
--- a/include/libs/sync/sync.h
+++ b/include/libs/sync/sync.h
@@ -138,6 +138,7 @@ typedef struct SSyncFSM {
   void (*FpRestoreFinishCb)(const struct SSyncFSM* pFsm);
   void (*FpReConfigCb)(const struct SSyncFSM* pFsm, const SRpcMsg* pMsg, const SReConfigCbMeta* pMeta);
   void (*FpLeaderTransferCb)(const struct SSyncFSM* pFsm, const SRpcMsg* pMsg, const SFsmCbMeta* pMeta);
+  bool (*FpApplyQueueEmptyCb)(const struct SSyncFSM* pFsm);
 
   void (*FpBecomeLeaderCb)(const struct SSyncFSM* pFsm);
   void (*FpBecomeFollowerCb)(const struct SSyncFSM* pFsm);
diff --git a/source/dnode/mnode/impl/src/mndSync.c b/source/dnode/mnode/impl/src/mndSync.c
index 0f50391ac5..e6e134c6a1 100644
--- a/source/dnode/mnode/impl/src/mndSync.c
+++ b/source/dnode/mnode/impl/src/mndSync.c
@@ -202,6 +202,13 @@ static void mndBecomeLeader(const SSyncFSM *pFsm) {
   SMnode *pMnode = pFsm->data;
 }
 
+static bool mndApplyQueueEmpty(const SSyncFSM *pFsm) {
+  SMnode *pMnode = pFsm->data;
+
+  int32_t itemSize = tmsgGetQueueSize(&pMnode->msgCb, 1, APPLY_QUEUE);
+  return (itemSize == 0);
+}
+
 SSyncFSM *mndSyncMakeFsm(SMnode *pMnode) {
   SSyncFSM *pFsm = taosMemoryCalloc(1, sizeof(SSyncFSM));
   pFsm->data = pMnode;
@@ -210,6 +217,7 @@ SSyncFSM *mndSyncMakeFsm(SMnode *pMnode) {
   pFsm->FpRollBackCb = NULL;
   pFsm->FpRestoreFinishCb = mndRestoreFinish;
   pFsm->FpLeaderTransferCb = NULL;
+  pFsm->FpApplyQueueEmptyCb = mndApplyQueueEmpty;
   pFsm->FpReConfigCb = NULL;
   pFsm->FpBecomeLeaderCb = mndBecomeLeader;
   pFsm->FpBecomeFollowerCb = mndBecomeFollower;
diff --git a/source/dnode/vnode/src/vnd/vnodeSync.c b/source/dnode/vnode/src/vnd/vnodeSync.c
index 3913561ae7..e7f8c9f562 100644
--- a/source/dnode/vnode/src/vnd/vnodeSync.c
+++ b/source/dnode/vnode/src/vnd/vnodeSync.c
@@ -436,6 +436,12 @@ static void vnodeBecomeLeader(const SSyncFSM *pFsm) {
   vDebug("vgId:%d, become leader", pVnode->config.vgId);
 }
 
+static bool vnodeApplyQueueEmpty(const SSyncFSM *pFsm) {
+  SVnode *pVnode = pFsm->data;
+  int32_t itemSize = tmsgGetQueueSize(&pVnode->msgCb, pVnode->config.vgId, APPLY_QUEUE);
+  return (itemSize == 0);
+}
+
 static SSyncFSM *vnodeSyncMakeFsm(SVnode *pVnode) {
   SSyncFSM *pFsm = taosMemoryCalloc(1, sizeof(SSyncFSM));
   pFsm->data = pVnode;
@@ -445,6 +451,7 @@ static SSyncFSM *vnodeSyncMakeFsm(SVnode *pVnode) {
   pFsm->FpGetSnapshotInfo = vnodeSyncGetSnapshot;
   pFsm->FpRestoreFinishCb = vnodeRestoreFinish;
   pFsm->FpLeaderTransferCb = NULL;
+  pFsm->FpApplyQueueEmptyCb = vnodeApplyQueueEmpty;
   pFsm->FpBecomeLeaderCb = vnodeBecomeLeader;
   pFsm->FpBecomeFollowerCb = vnodeBecomeFollower;
   pFsm->FpReConfigCb = NULL;
diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c
index 47b8614717..38cc9f3dfe 100644
--- a/source/libs/sync/src/syncMain.c
+++ b/source/libs/sync/src/syncMain.c
@@ -452,16 +452,22 @@ bool syncIsReadyForRead(int64_t rid) {
 
   bool ready = false;
   if (pSyncNode->state == TAOS_SYNC_STATE_LEADER && !pSyncNode->restoreFinish) {
-    if (!pSyncNode->pLogStore->syncLogIsEmpty(pSyncNode->pLogStore)) {
-      SSyncRaftEntry* pEntry = NULL;
-      int32_t         code = pSyncNode->pLogStore->syncLogGetEntry(
-          pSyncNode->pLogStore, pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore), &pEntry);
-      if (code == 0 && pEntry != NULL) {
-        if (pEntry->originalRpcType == TDMT_SYNC_NOOP && pEntry->term == pSyncNode->pRaftStore->currentTerm) {
-          ready = true;
-        }
+    if (!pSyncNode->pFsm->FpApplyQueueEmptyCb(pSyncNode->pFsm)) {
+      // apply queue not empty
+      ready = false;
+
+    } else {
+      if (!pSyncNode->pLogStore->syncLogIsEmpty(pSyncNode->pLogStore)) {
+        SSyncRaftEntry* pEntry = NULL;
+        int32_t         code = pSyncNode->pLogStore->syncLogGetEntry(
+            pSyncNode->pLogStore, pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore), &pEntry);
+        if (code == 0 && pEntry != NULL) {
+          if (pEntry->originalRpcType == TDMT_SYNC_NOOP && pEntry->term == pSyncNode->pRaftStore->currentTerm) {
+            ready = true;
+          }
 
-        syncEntryDestory(pEntry);
+          syncEntryDestory(pEntry);
+        }
       }
     }
   }
-- 
GitLab