Merge remote-tracking branch 'origin/feature/3.0_mhli' into fix/TD-20798

include/libs/sync/sync.h

@@ -211,9 +211,14 @@ typedef struct SSyncInfo {
   int32_t (*syncEqCtrlMsg)(const SMsgCb* msgcb, SRpcMsg* pMsg);
 } SSyncInfo;
 
+// if state == leader
+//     if restored, display "leader"
+//     if !restored && canRead, display "leader*"
+//     if !restored && !canRead, display "leader**"
 typedef struct SSyncState {
   ESyncState state;
   bool       restored;
+  bool       canRead;
 } SSyncState;
 
 int32_t syncInit();

source/libs/sync/inc/syncInt.h

@@ -244,6 +244,7 @@ int32_t   syncCacheEntry(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry, LRUHa
 bool      syncNodeHeartbeatReplyTimeout(SSyncNode* pSyncNode);
 bool      syncNodeSnapshotSending(SSyncNode* pSyncNode);
 bool      syncNodeSnapshotRecving(SSyncNode* pSyncNode);
+bool      syncNodeIsReadyForRead(SSyncNode* pSyncNode);
 
 // raft state change --------------
 void syncNodeUpdateTerm(SSyncNode* pSyncNode, SyncTerm term);

source/libs/sync/src/syncMain.c

@@ -381,15 +381,13 @@ int32_t syncStepDown(int64_t rid, SyncTerm newTerm) {
   return 0;
 }
 
-bool syncIsReadyForRead(int64_t rid) {
-  SSyncNode* pSyncNode = syncNodeAcquire(rid);
+bool syncNodeIsReadyForRead(SSyncNode* pSyncNode) {
   if (pSyncNode == NULL) {
     sError("sync ready for read error");
     return false;
   }
 
   if (pSyncNode->state == TAOS_SYNC_STATE_LEADER && pSyncNode->restoreFinish) {
-    syncNodeRelease(pSyncNode);
     return true;
   }
 
@@ -443,6 +441,18 @@ bool syncIsReadyForRead(int64_t rid) {
     }
   }
 
+  return ready;
+}
+
+bool syncIsReadyForRead(int64_t rid) {
+  SSyncNode* pSyncNode = syncNodeAcquire(rid);
+  if (pSyncNode == NULL) {
+    sError("sync ready for read error");
+    return false;
+  }
+
+  bool ready = syncNodeIsReadyForRead(pSyncNode);
+
   syncNodeRelease(pSyncNode);
   return ready;
 }
@@ -521,6 +531,7 @@ SSyncState syncGetState(int64_t rid) {
   if (pSyncNode != NULL) {
     state.state = pSyncNode->state;
     state.restored = pSyncNode->restoreFinish;
+    state.canRead = syncNodeIsReadyForRead(pSyncNode);
     syncNodeRelease(pSyncNode);
   }
 

source/libs/sync/src/syncSnapshot.c

@@ -541,6 +541,10 @@ _START_RECEIVER:
       taosMsleep(10);
     }
 
+    if (snapshotReceiverIsStart(pReceiver)) {
+      snapshotReceiverForceStop(pReceiver);
+    }
+
     snapshotReceiverStart(pReceiver, pMsg);  // set start-time same with sender
   }