Merge pull request #18716 from taosdata/feature/3.0_sync_pipeline

fix(sync): close snapshot receiver in pre-close

source/dnode/vnode/src/vnd/vnodeSync.c

@@ -606,16 +606,7 @@ void vnodeSyncPreClose(SVnode *pVnode) {
   vInfo("vgId:%d, pre close sync", pVnode->config.vgId);
   syncLeaderTransfer(pVnode->sync);
   syncPreStop(pVnode->sync);
-#if 0
-  while (syncSnapshotRecving(pVnode->sync)) {
-    vInfo("vgId:%d, snapshot is recving", pVnode->config.vgId);
-    taosMsleep(300);
-  }
-  while (syncSnapshotSending(pVnode->sync)) {
-    vInfo("vgId:%d, snapshot is sending", pVnode->config.vgId);
-    taosMsleep(300);
-  }
-#endif
+
   taosThreadMutexLock(&pVnode->lock);
   if (pVnode->blocked) {
     vInfo("vgId:%d, post block after close sync", pVnode->config.vgId);

source/libs/sync/src/syncMain.c

@@ -1221,6 +1221,26 @@ int32_t syncNodeStartStandBy(SSyncNode* pSyncNode) {
 }
 
 void syncNodePreClose(SSyncNode* pSyncNode) {
+  if (pSyncNode != NULL && pSyncNode->pFsm != NULL && pSyncNode->pFsm->FpApplyQueueItems != NULL) {
+    while (1) {
+      int32_t aqItems = pSyncNode->pFsm->FpApplyQueueItems(pSyncNode->pFsm);
+      sTrace("vgId:%d, pre close, %d items in apply queue", pSyncNode->vgId, aqItems);
+      if (aqItems == 0 || aqItems == -1) {
+        break;
+      }
+      taosMsleep(20);
+    }
+  }
+
+  if (pSyncNode->pNewNodeReceiver != NULL) {
+    if (snapshotReceiverIsStart(pSyncNode->pNewNodeReceiver)) {
+      snapshotReceiverForceStop(pSyncNode->pNewNodeReceiver);
+    }
+
+    snapshotReceiverDestroy(pSyncNode->pNewNodeReceiver);
+    pSyncNode->pNewNodeReceiver = NULL;
+  }
+
   // stop elect timer
   syncNodeStopElectTimer(pSyncNode);