fix: drop db has stream exist

source/dnode/mgmt/mgmt_vnode/src/vmInt.c

@@ -21,7 +21,7 @@ SVnodeObj *vmAcquireVnode(SVnodeMgmt *pMgmt, int32_t vgId) {
 
   taosThreadRwlockRdlock(&pMgmt->lock);
   taosHashGetDup(pMgmt->hash, &vgId, sizeof(int32_t), (void *)&pVnode);
-  if (pVnode == NULL) {
+  if (pVnode == NULL || pVnode->dropped) {
     terrno = TSDB_CODE_VND_INVALID_VGROUP_ID;
   } else {
     int32_t refCount = atomic_add_fetch_32(&pVnode->refCount, 1);
@@ -81,16 +81,18 @@ void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
 
   vmReleaseVnode(pMgmt, pVnode);
   while (pVnode->refCount > 0) taosMsleep(10);
+  dTrace("vgId:%d, wait for vnode queue is empty", pVnode->vgId);
+
   while (!taosQueueEmpty(pVnode->pWriteQ)) taosMsleep(10);
   while (!taosQueueEmpty(pVnode->pSyncQ)) taosMsleep(10);
   while (!taosQueueEmpty(pVnode->pApplyQ)) taosMsleep(10);
   while (!taosQueueEmpty(pVnode->pQueryQ)) taosMsleep(10);
   while (!taosQueueEmpty(pVnode->pFetchQ)) taosMsleep(10);
+  dTrace("vgId:%d, vnode-fetch queue is empty", pVnode->vgId);
 
   vmFreeQueue(pMgmt, pVnode);
   vnodeClose(pVnode->pImpl);
   pVnode->pImpl = NULL;
-
   dDebug("vgId:%d, vnode is closed", pVnode->vgId);
 
   if (pVnode->dropped) {

source/dnode/mgmt/mgmt_vnode/src/vmWorker.c

@@ -107,7 +107,7 @@ static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOf
     const STraceId *trace = &pMsg->info.traceId;
     dGTrace("vgId:%d, msg:%p get from vnode-sync queue", pVnode->vgId, pMsg);
 
-    int32_t code = vnodeProcessSyncReq(pVnode->pImpl, pMsg, NULL); // no response here
+    int32_t code = vnodeProcessSyncReq(pVnode->pImpl, pMsg, NULL);  // no response here
     dGTrace("vgId:%d, msg:%p is freed, code:0x%x", pVnode->vgId, pMsg, code);
     rpcFreeCont(pMsg->pCont);
     taosFreeQitem(pMsg);
@@ -146,8 +146,8 @@ static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtyp
 
   SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId);
   if (pVnode == NULL) {
-    dGError("vgId:%d, msg:%p failed to put into vnode queue since %s, type:%s", pHead->vgId, pMsg, terrstr(),
-            TMSG_INFO(pMsg->msgType));
+    dGError("vgId:%d, msg:%p failed to put into vnode queue since %s, msgtype:%s qtype:%d", pHead->vgId, pMsg,
+            terrstr(), TMSG_INFO(pMsg->msgType), qtype);
     return terrno != 0 ? terrno : -1;
   }
 

source/dnode/mnode/impl/inc/mndStream.h

@@ -38,6 +38,8 @@ int32_t mndPersistStream(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream);
 int32_t mndDropStreamTasks(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream);
 int32_t mndPersistDropStreamLog(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream);
 
+int32_t mndDropStreamByDb(SMnode *pMnode, STrans *pTrans, SDbObj *pDb);
+
 #ifdef __cplusplus
 }
 #endif

source/dnode/mnode/impl/src/mndDb.c

@@ -21,6 +21,7 @@
 #include "mndShow.h"
 #include "mndSma.h"
 #include "mndStb.h"
+#include "mndStream.h"
 #include "mndSubscribe.h"
 #include "mndTopic.h"
 #include "mndTrans.h"
@@ -927,6 +928,7 @@ static int32_t mndDropDb(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb) {
   if (mndDropOffsetByDB(pMnode, pTrans, pDb) != 0) goto _OVER;
   if (mndDropSubByDB(pMnode, pTrans, pDb) != 0) goto _OVER;
   if (mndDropTopicByDB(pMnode, pTrans, pDb) != 0) goto _OVER;
+  if (mndDropStreamByDb(pMnode, pTrans, pDb) != 0) goto _OVER;
   if (mndDropSmasByDb(pMnode, pTrans, pDb) != 0) goto _OVER;
   if (mndSetDropDbRedoActions(pMnode, pTrans, pDb) != 0) goto _OVER;
 
@@ -947,7 +949,6 @@ static int32_t mndDropDb(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb) {
   mndTransSetRpcRsp(pTrans, pRsp, rspLen);
 
   if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;
-
   code = 0;
 
 _OVER:

source/dnode/mnode/impl/src/mndSma.c

@@ -523,6 +523,7 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
   streamObj.updateTime = streamObj.createTime;
   streamObj.uid = mndGenerateUid(pCreate->name, strlen(pCreate->name));
   streamObj.sourceDbUid = pDb->uid;
+  streamObj.targetDbUid = pDb->uid;
   streamObj.version = 1;
   streamObj.sql = pCreate->sql;
   streamObj.smaId = smaObj.uid;
@@ -853,51 +854,26 @@ _OVER:
 }
 
 int32_t mndDropSmasByDb(SMnode *pMnode, STrans *pTrans, SDbObj *pDb) {
-  SSdb    *pSdb = pMnode->pSdb;
-  SSmaObj *pSma = NULL;
-  void    *pIter = NULL;
-  SVgObj  *pVgroup = NULL;
-  int32_t  code = -1;
+  SSdb *pSdb = pMnode->pSdb;
+  void *pIter = NULL;
 
   while (1) {
+    SSmaObj *pSma = NULL;
     pIter = sdbFetch(pSdb, SDB_SMA, pIter, (void **)&pSma);
     if (pIter == NULL) break;
 
     if (pSma->dbUid == pDb->uid) {
-      pVgroup = mndAcquireVgroup(pMnode, pSma->dstVgId);
-      if (pVgroup == NULL) goto _OVER;
-
-      pVgroup = mndAcquireVgroup(pMnode, pSma->dstVgId);
-      if (pVgroup == NULL) goto _OVER;
-
-      SStreamObj *pStream = mndAcquireStream(pMnode, pSma->name);
-      if (pStream != NULL && pStream->smaId == pSma->uid) {
-        if (mndDropStreamTasks(pMnode, pTrans, pStream) < 0) {
-          mError("stream:%s, failed to drop task since %s", pStream->name, terrstr());
-          goto _OVER;
-        }
-        if (mndPersistDropStreamLog(pMnode, pTrans, pStream) < 0) {
-          goto _OVER;
-        }
+      if (mndSetDropSmaCommitLogs(pMnode, pTrans, pSma) != 0) {
+        sdbRelease(pSdb, pSma);
+        sdbCancelFetch(pSdb, pSma);
+        return -1;
       }
-
-      if (mndSetDropSmaVgroupCommitLogs(pMnode, pTrans, pVgroup) != 0) goto _OVER;
-      if (mndSetDropSmaVgroupRedoActions(pMnode, pTrans, pDb, pVgroup) != 0) goto _OVER;
-      if (mndSetDropSmaCommitLogs(pMnode, pTrans, pSma) != 0) goto _OVER;
-      mndReleaseVgroup(pMnode, pVgroup);
-      pVgroup = NULL;
     }
 
     sdbRelease(pSdb, pSma);
   }
 
-  code = 0;
-
-_OVER:
-  sdbCancelFetch(pSdb, pIter);
-  sdbRelease(pSdb, pSma);
-  mndReleaseVgroup(pMnode, pVgroup);
-  return code;
+  return 0;
 }
 
 static int32_t mndProcessDropSmaReq(SRpcMsg *pReq) {

source/dnode/mnode/impl/src/mndStream.c

@@ -673,27 +673,29 @@ static int32_t mndProcessDropStreamReq(SRpcMsg *pReq) {
 
 int32_t mndDropStreamByDb(SMnode *pMnode, STrans *pTrans, SDbObj *pDb) {
   SSdb *pSdb = pMnode->pSdb;
+  void *pIter = NULL;
 
-  void       *pIter = NULL;
-  SStreamObj *pStream = NULL;
   while (1) {
+    SStreamObj *pStream = NULL;
     pIter = sdbFetch(pSdb, SDB_STREAM, pIter, (void **)&pStream);
     if (pIter == NULL) break;
 
     if (pStream->sourceDbUid == pDb->uid || pStream->targetDbUid == pDb->uid) {
       if (pStream->sourceDbUid != pStream->targetDbUid) {
         sdbRelease(pSdb, pStream);
+        sdbCancelFetch(pSdb, pIter);
+        mError("db:%s, failed to drop stream:%s since sourceDbUid:%" PRId64 " not match with targetDbUid:%" PRId64,
+               pDb->name, pStream->name, pStream->sourceDbUid, pStream->targetDbUid);
+        terrno = TSDB_CODE_MND_STREAM_ALREADY_EXIST;
         return -1;
       } else {
         // TODO drop all task on snode
         if (mndPersistDropStreamLog(pMnode, pTrans, pStream) < 0) {
           sdbRelease(pSdb, pStream);
+          sdbCancelFetch(pSdb, pIter);
           return -1;
         }
       }
-    } else {
-      sdbRelease(pSdb, pStream);
-      continue;
     }
 
 #if 0

source/dnode/mnode/impl/src/mndTrans.c

@@ -90,7 +90,7 @@ static int32_t mndTransGetActionsSize(SArray *pArray) {
   for (int32_t i = 0; i < actionNum; ++i) {
     STransAction *pAction = taosArrayGet(pArray, i);
     if (pAction->actionType == TRANS_ACTION_RAW) {
-      rawDataLen += (sdbGetRawTotalSize(pAction->pRaw) + sizeof(int32_t));
+      rawDataLen += (sizeof(STransAction) + sdbGetRawTotalSize(pAction->pRaw));
     } else if (pAction->actionType == TRANS_ACTION_MSG) {
       rawDataLen += (sizeof(STransAction) + pAction->contLen);
     } else {
@@ -105,7 +105,7 @@ static int32_t mndTransGetActionsSize(SArray *pArray) {
 static SSdbRaw *mndTransActionEncode(STrans *pTrans) {
   terrno = TSDB_CODE_OUT_OF_MEMORY;
 
-  int32_t rawDataLen = sizeof(STrans) + TRANS_RESERVE_SIZE;
+  int32_t rawDataLen = sizeof(STrans) + TRANS_RESERVE_SIZE + pTrans->paramLen;
   rawDataLen += mndTransGetActionsSize(pTrans->redoActions);
   rawDataLen += mndTransGetActionsSize(pTrans->undoActions);
   rawDataLen += mndTransGetActionsSize(pTrans->commitActions);
@@ -226,7 +226,8 @@ static SSdbRaw *mndTransActionEncode(STrans *pTrans) {
 
 _OVER:
   if (terrno != 0) {
-    mError("trans:%d, failed to encode to raw:%p len:%d since %s", pTrans->id, pRaw, dataPos, terrstr());
+    mError("trans:%d, failed to encode to raw:%p maxlen:%d len:%d since %s", pTrans->id, pRaw, sdbGetRawTotalSize(pRaw),
+           dataPos, terrstr());
     sdbFreeRaw(pRaw);
     return NULL;
   }

source/dnode/vnode/src/tq/tqSink.c

@@ -185,5 +185,7 @@ void tqTableSink(SStreamTask* pTask, void* vnode, int64_t ver, void* data) {
       .contLen = ntohl(pReq->length),
   };
 
-  ASSERT(tmsgPutToQueue(&pVnode->msgCb, WRITE_QUEUE, &msg) == 0);
+  if (tmsgPutToQueue(&pVnode->msgCb, WRITE_QUEUE, &msg) != 0) {
+    tqDebug("failed to put into write-queue since %s", terrstr());
+  }
 }

tests/script/jenkins/basic.txt

@@ -93,6 +93,7 @@
 ./test.sh -f tsim/stream/basic0.sim
 ./test.sh -f tsim/stream/basic1.sim
 ./test.sh -f tsim/stream/basic2.sim
+./test.sh -f tsim/stream/drop_stream.sim
 ./test.sh -f tsim/stream/distributeInterval0.sim
 # ./test.sh -f tsim/stream/distributeIntervalRetrive0.sim
 # ./test.sh -f tsim/stream/distributesession0.sim